Codex 智能体实战：从零系统学习智能体应用

┌─────────────────────────────────────────────┐
│              智能体系统架构                    │
├─────────────────────────────────────────────┤
│  ┌───────────┐                             │
│  │  用户输入  │                             │
│  └─────┬─────┘                             │
│        ▼                                     │
│  ┌───────────┐  ┌───────────┐              │
│  │  规划模块  │◄─┤  记忆模块  │              │
│  │ (拆解任务) │  │ (短期/长期)│              │
│  └─────┬─────┘  └───────────┘              │
│        ▼                                     │
│  ┌───────────┐  ┌───────────┐              │
│  │  执行模块  │──►│  工具调用  │              │
│  │ (调用工具) │  │ (API/代码) │              │
│  └─────┬─────┘  └───────────┘              │
│        ▼                                     │
│  ┌───────────┐                             │
│  │  反思模块  │  ← 检查结果，决定下一步      │
│  └───────────┘                             │
└─────────────────────────────────────────────┘

# 1. 创建项目目录
mkdir codex-agent-tutorial && cd codex-agent-tutorial

# 2. 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows 用 venv\Scripts\activate

# 3. 安装依赖
pip install langchain langchain-openai python-dotenv

# 4. 创建 .env 文件（填入你的 API Key）
echo "OPENAI_API_KEY=sk-xxx" > .env

# 5. 创建主文件
touch agent.py

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain.prompts import PromptTemplate

# 加载环境变量
load_dotenv()

# 1. 初始化 Codex 模型（通过 OpenAI API）
llm = ChatOpenAI(
    model="gpt-4o",  # 或使用 gpt-4o-codex 专用模型
    temperature=0.1,
    api_key=os.getenv("OPENAI_API_KEY")
)

# 2. 定义一个简单工具
@tool
def get_current_time() -> str:
    """获取当前日期和时间。"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 3. 构建智能体
tools = [get_current_time]

prompt = PromptTemplate.from_template("""
你是一个有用的AI助手。你可以使用以下工具来回答问题：

{tools}

工具名称: {tool_names}

用户问题: {input}

请一步步思考，如果需要使用工具就调用，否则直接回答。

思考过程: {agent_scratchpad}
""")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 打印思考过程，便于学习
    max_iterations=5  # 防止无限循环
)

# 4. 运行智能体
if __name__ == "__main__":
    result = agent_executor.invoke({
        "input": "现在几点了？"
    })
    print(f"答案: {result['output']}")

python agent.py

> Entering new AgentExecutor chain...
我需要获取当前时间，调用 get_current_time 工具。
Action: get_current_time
Action Input: 
Observation: 2026-07-08 14:35:22
Thought: 我已经获取到当前时间，可以回答用户了。
Final Answer: 现在是 2026年7月8日 14:35:22。

> Finished chain.
答案: 现在是 2026年7月8日 14:35:22。

from langchain.tools import tool

# 方式一：装饰器（推荐，最简单）
@tool
def multiply(a: float, b: float) -> float:
    """两个数相乘。"""
    return a * b

# 方式二：显式定义（更灵活）
from langchain.tools import StructuredTool

def search_web(query: str) -> str:
    # 模拟搜索
    return f"关于'{query}'的搜索结果：..."

search_tool = StructuredTool.from_function(
    func=search_web,
    name="web_search",
    description="搜索互联网获取信息，参数为搜索关键词"
)

import subprocess
import json
import requests

@tool
def read_file(filepath: str) -> str:
    """读取文件内容。"""
    with open(filepath, 'r', encoding='utf-8') as f:
        return f.read()

@tool
def write_file(filepath: str, content: str) -> str:
    """写入内容到文件。"""
    with open(filepath, 'w', encoding='utf-8') as f:
        f.write(content)
    return f"文件 {filepath} 写入成功"

@tool
def execute_command(command: str) -> str:
    """执行命令行命令（谨慎使用）。"""
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout if result.returncode == 0 else result.stderr

@tool
def search_github(repo_name: str) -> str:
    """在 GitHub 上搜索仓库。"""
    url = f"https://api.github.com/search/repositories?q={repo_name}"
    response = requests.get(url)
    data = response.json()
    top_repos = data.get('items', [])[:3]
    return json.dumps([{
        'name': r['full_name'],
        'stars': r['stargazers_count'],
        'description': r['description']
    } for r in top_repos], ensure_ascii=False)

# 组装工具集
developer_tools = [read_file, write_file, execute_command, search_github]

from langchain.memory import ConversationBufferMemory
from langchain.agents import initialize_agent

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent_with_memory = initialize_agent(
    tools=tools,
    llm=llm,
    agent="chat-conversational-react-description",
    memory=memory,
    verbose=True
)

# 多轮对话测试
print(agent_with_memory.run("我叫小明"))
print(agent_with_memory.run("我叫什么名字？"))  # 会记住上轮回答

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader

# 1. 加载文档（你的"私密知识库"）
loader = TextLoader("my_knowledge.txt")
documents = loader.load()

# 2. 分块
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(documents)

# 3. 向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)

# 4. 检索函数
@tool
def search_knowledge(query: str) -> str:
    """从知识库中检索信息。"""
    docs = vectorstore.similarity_search(query, k=3)
    return "\n".join([d.page_content for d in docs])

# 5. 添加到智能体
agent_tools = developer_tools + [search_knowledge]

from langchain.chains import LLMChain

planner_prompt = """
你是一个任务规划专家。请将以下用户目标拆解为 3-5 个可执行的子任务，每个子任务描述清楚需要做什么、调用什么工具。

目标: {goal}

输出格式（JSON）：
{{
    "goal": "原始目标",
    "subtasks": [
        {{"id": 1, "description": "...", "tools_needed": ["..."]}},
        ...
    ]
}}
"""

planner_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate.from_template(planner_prompt)
)

# 使用示例
plan = planner_chain.invoke({"goal": "帮我给项目添加一个 README.md，包含安装步骤和使用示例"})
print(plan)

@tool
def execute_with_retry(code: str, max_retries: int = 3) -> str:
    """执行代码，失败后自动修正并重试。"""
    for attempt in range(max_retries):
        try:
            result = exec_compiled_code(code)  # 假想的执行函数
            return f"执行成功！结果: {result}"
        except Exception as e:
            if attempt == max_retries - 1:
                return f"执行失败（已重试{max_retries}次）: {str(e)}"
            # 让 LLM 修正代码
            code = self._fix_code(code, str(e))
    return "执行失败"

import os
import json
from github import Github
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# ==================== 配置 ====================
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
REPO_NAME = "your-username/your-repo"

# ==================== 工具定义 ====================

@tool
def get_pr_files(pr_number: int) -> str:
    """获取 PR 中所有变更的文件列表及 diff 内容。"""
    g = Github(GITHUB_TOKEN)
    repo = g.get_repo(REPO_NAME)
    pr = repo.get_pull(pr_number)
    
    files_info = []
    for file in pr.get_files():
        files_info.append({
            "filename": file.filename,
            "status": file.status,
            "additions": file.additions,
            "deletions": file.deletions,
            "patch": file.patch  # 具体变更内容
        })
    return json.dumps(files_info, ensure_ascii=False)

@tool
def analyze_code_quality(file_content: str) -> str:
    """分析一段代码的质量，返回潜在问题和改进建议。"""
    # 这里可以集成 SonarQube / ESLint 等静态分析工具
    # 简化版：让 LLM 自行分析
    pass

@tool
def generate_review_report(issues: list) -> str:
    """根据发现的 issues 生成 Markdown 格式的审查报告。"""
    report = f"# 代码审查报告\n\n"
    report += f"## 概要\n- 共发现 {len(issues)} 个问题\n\n"
    report += f"## 详情\n\n"
    for idx, issue in enumerate(issues, 1):
        report += f"### {idx}. {issue['filename']}: {issue['severity']}\n"
        report += f"- 问题描述: {issue['description']}\n"
        report += f"- 建议修改: {issue['suggestion']}\n\n"
    return report

# ==================== 智能体创建 ====================

def create_code_review_agent():
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    
    tools = [get_pr_files, generate_review_report]
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个资深代码审查专家。你的任务是：
        1. 分析 PR 中的代码变更
        2. 发现潜在的 Bug、安全漏洞、性能问题
        3. 给出具体的改进建议
        4. 生成结构化的审查报告
        
        请重点关注：
        - 空指针 / 未处理异常
        - SQL 注入 / XSS 等安全问题
        - 性能瓶颈（循环、数据库查询）
        - 代码规范和可读性
        """),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])
    
    agent = create_openai_tools_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)

# ==================== 运行 ====================

if __name__ == "__main__":
    agent = create_code_review_agent()
    result = agent.invoke({
        "input": "请审查 PR #42 的代码变更，生成完整的审查报告"
    })
    print(result['output'])

import gradio as gr

def chat_with_agent(message, history):
    response = agent_executor.invoke({"input": message})
    return response['output']

interface = gr.ChatInterface(
    fn=chat_with_agent,
    title="Codex 智能体助手",
    description="基于 GPT-4o 构建的编程智能体"
)

interface.launch(share=True)  # share=True 生成公网链接