LangChain与AutoGPT思想融合：手写Agent核心循环与记忆机制

langchain==0.1.0
langchain-openai==0.0.2.post1
pydantic==1.10.13  # 必须锁1.x版本！
openai==1.6.1
python==3.10.11

┌─────────────────────────────────────┐
│         Agent核心循环 (while loop)     │
├─────────────────────────────────────┤
│  工具注册表│记忆堆栈│错误修正器│循环控制  │
├─────────────────────────────────────┤
│      OpenAI ChatCompletion API       │
└─────────────────────────────────────┘

from typing import Dict, Callable, Any
import json

# 工具函数签名规范：接收一个字符串参数，返回一个字符串
def query_order(order_id: str) -> str:
    """查询订单状态，order_id是订单号字符串"""
    # 这里模拟真实API调用，实际项目中替换为requests.post
    import time
    time.sleep(0.3)  # 模拟网络延迟
    return json.dumps({"status": "shipped", "eta": "2024-03-20"})

def create_return_request(order_id: str) -> str:
    """创建退货申请，返回退单号"""
    return json.dumps({"return_id": "R12345", "status": "created"})

# 工具注册表：包含描述信息供LLM理解
TOOLS: Dict[str, Dict[str, Any]] = {
    "query_order": {
        "func": query_order,
        "description": "查询订单状态。输入应为订单号（数字字符串）。当用户提到'订单'、'快递'时使用。",
    },
    "create_return_request": {
        "func": create_return_request,
        "description": "创建退货申请。输入应为订单号。当用户明确表示要'退货'、'退款'时使用。",
    }
}

def call_tool(tool_name: str, arg: str) -> str:
    """统一工具调用入口，带异常捕获"""
    tool = TOOLS.get(tool_name)
    if not tool:
        return f"错误：未知工具{tool_name}"
    try:
        return tool["func"](arg)
    except Exception as e:
        return f"工具执行异常: {str(e)}"

from collections import deque

class MemoryStack:
    def __init__(self, max_len: int = 5):
        self.messages = deque(maxlen=max_len)

    def add_user(self, content: str):
        self.messages.append({"role": "user", "content": content})

    def add_assistant(self, content: str):
        self.messages.append({"role": "assistant", "content": content})

    def add_tool_result(self, content: str):
        # 工具结果作为assistant消息的一部分，但加前缀标记
        self.messages.append({"role": "assistant", "content": f"[工具结果] {content}"})

    def get_context(self) -> list:
        return list(self.messages)

SYSTEM_PROMPT = """你是一个电商客服助手。请根据对话历史，输出一个JSON决策。
格式：
{"thought": "你的思考", "tool": "工具名或finish", "arg": "参数", "reply": "给用户的回复"}
规则：
1. 如果用户需要查询订单，设置tool为"query_order"，arg为订单号
2. 如果用户要退货，设置tool为"create_return_request"
3. 如果对话结束或无需工具，设置tool为"finish"
4. 回复要友好且简洁。"""

def run_agent(user_input: str, max_iterations: int = 5) -> str:
    memory = MemoryStack()
    memory.add_user(user_input)

    for i in range(max_iterations):
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-1106",
            messages=[{"role": "system", "content": SYSTEM_PROMPT}] + memory.get_context(),
            temperature=0.2,
            max_tokens=300,
            response_format={"type": "json_object"}  # 强制JSON输出
        )

        raw = response.choices[0].message.content
        # 错误处理1：JSON解析失败时，尝试提取JSON片段
        try:
            decision = json.loads(raw)
        except json.JSONDecodeError:
            import re
            match = re.search(r'\{.*\}', raw, re.DOTALL)
            if match:
                decision = json.loads(match.group())
            else:
                memory.add_assistant("抱歉，我没理解您的意思，请再说一遍？")
                continue

        # 错误处理2：工具名不存在时，强制改走finish
        if decision.get("tool") != "finish" and decision.get("tool") not in TOOLS:
            decision["tool"] = "finish"
            decision["reply"] = "我暂时无法处理这个请求，已转接人工。"

        # 执行工具
        if decision["tool"] == "finish":
            memory.add_assistant(decision["reply"])
            return decision["reply"]
        else:
            result = call_tool(decision["tool"], decision.get("arg", ""))
            memory.add_tool_result(result)
            # 把工具结果反馈给LLM，让LLM决定下一步

    return "抱歉，处理超时，请稍后重试。"