从零构建LangChain Agent：工具/记忆/容错全链路实现

# 必须精确版本，否则工具Schema解析会报错
langchain==0.3.17  
langchain-community==0.3.14  
langchain-openai==0.2.12  
openai==1.55.0  
httpx==0.28.1  
python-dotenv==1.0.1  

graph TD
    A[用户输入] --> B[AgentExecutor]
    B --> C{LLM推理}
    C -->|需要工具| D[工具列表]
    D --> E[爬虫工具]
    D --> F[API工具]  
    D --> G[分析工具]
    E --> H[错误处理模块]
    H -->|重试3次| E
    H -->|降级| I[缓存数据]
    C -->|完成| J[输出结果]
    B --> K[记忆管理]
    K -->|最近20轮| B

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, Optional
import httpx

class GitHubIssueInput(BaseModel):
    repo_url: str = Field(description="GitHub仓库完整URL，如https://github.com/owner/repo")
    max_issues: int = Field(default=10, ge=1, le=50, description="最多获取的issue数量")

class GitHubIssueTool(BaseTool):
    name = "github_issue_retriever"
    description = """
    从GitHub仓库获取最近的Issues，返回标题、状态、标签、创建时间。
    如果返回空列表，说明仓库可能不存在或无公开issue。
    注意：本工具最多重试3次，每次间隔2秒。
    """
    args_schema: Type[BaseModel] = GitHubIssueInput
    max_retries: int = 3
    retry_delay: float = 2.0

    def _run(self, repo_url: str, max_issues: int = 10) -> str:
        # 解析owner/repo
        parts = repo_url.rstrip("/").split("/")
        owner, repo = parts[-2], parts[-1]
        api_url = f"https://api.github.com/repos/{owner}/{repo}/issues?state=all&per_page={max_issues}"

        for attempt in range(self.max_retries):
            try:
                resp = httpx.get(api_url, headers={"Accept": "application/vnd.github.v3+json"}, timeout=15)
                resp.raise_for_status()
                issues = resp.json()
                # 格式化输出
                result = []
                for issue in issues[:max_issues]:
                    result.append(f"#{issue['number']} [{issue['state']}] {issue['title']} - {issue['created_at'][:10]}")
                return "\n".join(result) if result else "未找到任何issue"
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429 and attempt  str:
    error_str = str(error)
    # 三级错误处理
    if "429" in error_str:
        return "API限流，请等待5秒后重试当前步骤"
    elif "400" in error_str or "404" in error_str:
        return f"参数错误: {error_str}，尝试使用备用数据源"
    elif "Tool not found" in error_str:
        return "工具不存在，终止当前任务"
    else:
        return f"未知错误: {error_str}，跳过本步骤"

# 创建Agent
tools = [GitHubIssueTool(), ...]  # 其他工具类似实现
prompt = PromptTemplate.from_template("""
你是一个GitHub分析助手。请根据用户需求，逐步使用工具完成任务。
当前对话历史: {chat_history}
用户输入: {input}
可用工具: {tools}

请严格按照以下格式回答：
Thought: 你当前的思考
Action: 工具名称（从{tool_names}中选择）
Action Input: 工具输入（JSON格式）
Observation: 工具返回结果
...（可重复多轮）
Thought: 我已经完成所有操作
Final Answer: 最终回答
""")

agent = create_react_agent(
    llm=llm,
    tools=tools,
    prompt=prompt,
    # 关键：自定义输出解析器，处理LLM格式错误
    output_parser=...  # 使用默认的ReActSingleInputOutputParser
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    max_iterations=15,           # 防止无限循环
    max_execution_time=120,      # 整体超时2分钟
    handle_parsing_errors=custom_error_parser,  # 错误处理回调
    early_stopping_method="generate",  # 达到max_iterations时尝试生成答案
    verbose=True                 # 开发阶段开启调试日志
)