AI 智能体（Agent）架构设计：从 ReAct 到自主决策系统的完整实践

# 工具签名规范（OpenAI function calling 风格）
{
    "type": "function",
    "function": {
        "name": "search_flight",
        "description": "搜索指定日期和目的地的航班。",
        "parameters": {
            "type": "object",
            "properties": {
                "origin": {"type": "string", "description": "出发城市代码"},
                "destination": {"type": "string", "description": "到达城市代码"},
                "date": {"type": "string", "description": "出发日期(YYYY-MM-DD)"}
            },
            "required": ["origin", "destination", "date"]
        }
    }
}

你是智能助理 Agent，负责帮助用户完成目标。

可用工具：
- search_flight(origin, destination, date): 搜索航班
- compare_price(flight_list): 比较价格
- book_ticket(flight_id, passenger): 预订机票

规则：
1. 每次只能调用一个工具
2. 调用工具前，先用 <Thought> 说明你的推理过程
3. 工具返回后，用 <Observation> 记录结果
4. 任务完成时，直接回复用户

历史记录：
{{memory}}

用户目标：{{goal}}

import os, json, re, openai
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# ====== 工具函数 ======
def search_flight(origin: str, destination: str, date: str):
    """Mock 搜索航班"""
    return [
        {"id": "MU5161", "price": 580, "time": "08:00-10:30"},
        {"id": "CA1850", "price": 620, "time": "14:00-16:20"},
        {"id": "MU5133", "price": 690, "time": "19:00-21:30"},
    ]

def compare_price(flight_ids):
    return f"价格排序: MU5161(¥580) < CA1850(¥620) < MU5133(¥690)"

TOOLS = {
    "search_flight": search_flight,
    "compare_price": compare_price,
}

# ====== Agent 类 ======
class ReActAgent:
    def __init__(self, client):
        self.client = client
        self.memory = []
        self.max_loop = 10

    SYSTEM_PROMPT = """你是一个智能订票代理 Agent。每次只能执行一个操作。
可用工具：search_flight(origin, destination, date), compare_price(flight_ids)
你必须按格式回复：
Thought: [推理过程]
Action: 工具名称(参数) 或 "完成"
"""

    def call_tool(self, action_str):
        m = re.match(r'(\w+)\((.*)\)', action_str)
        if not m:
            return "格式错误"
        name, args = m.group(1), m.group(2)
        # 简单参数解析（生产环境请用 ast.literal_eval）
        try:
            arg_dict = eval(f"dict({args})")
            return TOOLS[name](**arg_dict)
        except Exception as e:
            return f"工具调用错误: {e}"

    def run(self, goal: str):
        self.memory.append({"role": "user", "content": goal})
        for _ in range(self.max_loop):
            messages = [
                {"role": "system", "content": self.SYSTEM_PROMPT},
                *self.memory,
            ]
            resp = self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                temperature=0.2,
            )
            content = resp.choices[0].message.content
            self.memory.append({"role": "assistant", "content": content})

            # 提取 Thought 和 Action
            thought = re.search(r'Thought:\s*(.*?)(?:\n|$)', content, re.DOTALL)
            action = re.search(r'Action:\s*(.*?)(?:\n|$)', content, re.DOTALL)

            if action:
                action_str = action.group(1).strip()
                if "完成" in action_str or not action_str:
                    print("✅ 任务完成")
                    return content
                # 调用工具
                result = self.call_tool(action_str)
                obs = f"Observation: {json.dumps(result, ensure_ascii=False)}"
                self.memory.append({"role": "user", "content": obs})
                print(f"🧠 Thought: {thought.group(1).strip() if thought else 'N/A'}")
                print(f"🔧 Action: {action_str}")
                print(f"📊 {obs}")
            else:
                break

# 运行
agent = ReActAgent(client)
result = agent.run("帮我查 2024-08-10 从上海到北京最便宜的机票")
print("\n=== 最终回复 ===\n", result)

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import Tool

# 初始化组件
llm = ChatOpenAI(model="gpt-4o-mini")
embedding = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embedding, persist_directory="./chroma_db")

# 创建检索工具
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
def retrieve_context(query):
    docs = retriever.invoke(query)
    return "\n".join([d.page_content for d in docs])

retrieval_tool = Tool.from_function(
    func=retrieve_context,
    name="knowledge_retrieval",
    description="从历史文档和公司知识库中检索相关信息。",
)

# Prompt 模板（带记忆插槽）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个客服 Agent。使用知识库回答用户问题。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# 创建 Agent
agent = create_openai_functions_agent(llm, [retrieval_tool], prompt)
executor = AgentExecutor(agent=agent, tools=[retrieval_tool], verbose=True)

# 运行
response = executor.invoke({
    "input": "我们公司支持哪些支付方式？",
    "chat_history": []
})