Agent构建方法详解

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# ① 系统提示：定义 Agent 身份和行为规范
system_prompt = """你是一个专业的研究助手，具备以下能力：
- 使用搜索工具查找最新信息
- 使用计算器进行数学计算
- 综合多源信息给出准确回答

始终遵循 ReAct 格式：先思考(Thought)，再行动(Action)，再观察(Observation)。
"""

# ② ChatPromptTemplate：结构化提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder(variable_name="chat_history"),  # 注入对话历史
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # Agent 推理过程
])

# ③ FewShot 提示：提供示例引导 LLM 行为
from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {"input": "今天北京天气怎么样？", 
     "output": "Thought: 需要查询实时天气\nAction: search\nAction Input: 北京今日天气"},
]

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=ChatPromptTemplate.from_messages([
        ("human", "{input}"), ("ai", "{output}")
    ]),
    examples=examples,
)

from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_experimental.tools import PythonREPLTool

# ① @tool 装饰器：最简单的工具定义方式
@tool
def calculate(expression: str) -> str:
    """计算数学表达式。输入应为有效的 Python 数学表达式，如 '2 + 3 * 4'"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {e}"

@tool  
def get_weather(city: str) -> str:
    """获取指定城市的当前天气信息。"""
    # 实际场景替换为真实 API
    return f"{city}：晴，25°C，东南风3级"

# ② 使用内置工具
search = DuckDuckGoSearchRun()
python_repl = PythonREPLTool()

# ③ 工具列表 — 注册给 Agent
tools = [calculate, get_weather, search, python_repl]

# ④ 查看工具 Schema（LLM 看到的工具描述）
print(calculate.name)         # "calculate"
print(calculate.description)  # "计算数学表达式..."
print(calculate.args_schema.schema())  # JSON Schema

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.messages import HumanMessage, AIMessage

# ① 初始化 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ② 创建 Agent（绑定工具和提示）
agent = create_openai_tools_agent(llm, tools, prompt)

# ③ AgentExecutor：负责循环执行直到任务完成
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # 打印推理过程
    max_iterations=10,      # 防止无限循环
    return_intermediate_steps=True,  # 返回中间步骤
    handle_parsing_errors=True,      # 优雅处理解析错误
)

# ④ 维护对话历史（多轮对话）
chat_history = []

def chat(user_input: str) -> str:
    response = agent_executor.invoke({
        "input": user_input,
        "chat_history": chat_history,
    })
    # 更新历史
    chat_history.extend([
        HumanMessage(content=user_input),
        AIMessage(content=response["output"]),
    ])
    return response["output"]

# ⑤ 运行示例
result = chat("帮我查一下今天上海的天气，然后换算成华氏度")
print(result)

# ====================================================
# LangChain ReAct Agent 完整原型
# pip install langchain langchain-openai duckduckgo-search
# ====================================================
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.agents import AgentExecutor, create_openai_tools_agent

os.environ["OPENAI_API_KEY"] = "your-api-key"

# --- 1. 定义工具 ---
@tool
def calculator(expr: str) -> str:
    """执行数学计算。示例：'100 * 1.13' 计算含税价格"""
    try:
        return str(round(eval(expr), 4))
    except:
        return "计算失败，请检查表达式"

search = DuckDuckGoSearchRun(name="web_search")
tools = [calculator, search]

# --- 2. 构建提示 ---
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个严谨的助手，擅长搜索信息和数学计算。"
               "分步骤思考，合理使用工具，给出准确答案。"),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

# --- 3. 创建 Agent ---
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent, tools=tools,
    verbose=True, max_iterations=8,
    handle_parsing_errors=True
)

# --- 4. 运行 ---
result = executor.invoke({"input": "特斯拉今天的股价是多少？如果我买100股需要多少钱？"})
print("\n📊 最终答案:", result["output"])

# ====================================================
# AutoGen 双 Agent 协作：AI助手 + 代码执行器
# pip install pyautogen
# ====================================================
import autogen

# LLM 配置
llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "your-key"}],
    "temperature": 0,
    "timeout": 120,
}

# --- 1. 定义 Assistant Agent（LLM 驱动，负责思考和生成代码）---
assistant = autogen.AssistantAgent(
    name="AI助手",
    system_message="""你是一个数据分析专家。
    当需要计算或处理数据时，编写 Python 代码。
    代码必须完整可运行，包含所有必要的 import。
    分析完成后说 'TERMINATE' 结束对话。""",
    llm_config=llm_config,
)

# --- 2. 定义 UserProxy Agent（代码执行器）---
user_proxy = autogen.UserProxyAgent(
    name="执行器",
    human_input_mode="NEVER",       # 完全自动，不需要人类介入
    max_consecutive_auto_reply=5,   # 最多自动回复5次
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
    code_execution_config={
        "work_dir": "workspace",    # 代码执行目录
        "use_docker": False,        # 本地执行（生产环境建议用 Docker）
    },
)

# --- 3. 启动对话 ---
user_proxy.initiate_chat(
    assistant,
    message="分析以下数字列表的统计特征并绘制直方图：[23,45,67,12,89,34,56,78,90,11,44,66]"
)

# ====================================================
# AutoGen 多 Agent 群聊：研究团队协作
# ====================================================
import autogen

llm_config = {"config_list": [{"model": "gpt-4o", "api_key": "your-key"}]}

# 定义三个专业角色
researcher = autogen.AssistantAgent(
    name="研究员",
    system_message="你负责收集和整理信息，提供客观的研究发现。",
    llm_config=llm_config,
)

critic = autogen.AssistantAgent(
    name="评论家",
    system_message="你负责批判性地审查信息，指出潜在问题和局限性。",
    llm_config=llm_config,
)

writer = autogen.AssistantAgent(
    name="写作者",
    system_message="你负责将研究结论整理成清晰、结构化的报告。完成后说 TERMINATE。",
    llm_config=llm_config,
)

user_proxy = autogen.UserProxyAgent(
    name="用户",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=1,
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
    code_execution_config=False,
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, critic, writer],
    messages=[],
    max_round=12,                      # 最多对话12轮
    speaker_selection_method="auto",   # 自动选择下一个发言人
)

manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# 启动
user_proxy.initiate_chat(
    manager,
    message="研究并分析：2025年 AI Agent 的主要应用场景和商业价值"
)

# ====================================================
# LangGraph ReAct Agent 完整实现
# pip install langgraph langchain-openai
# ====================================================
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator, json

# --- 1. 定义 State（图的全局状态）---
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]  # 消息自动追加

# --- 2. 定义工具 ---
@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    return f"[搜索结果] 关于'{query}'：这是模拟的搜索结果，实际请接入真实搜索API"

@tool
def run_python(code: str) -> str:
    """执行 Python 代码并返回结果"""
    try:
        import io, contextlib
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            exec(code, {})
        return output.getvalue() or "执行成功（无输出）"
    except Exception as e:
        return f"执行错误: {e}"

tools = [search_web, run_python]

# --- 3. 定义节点函数 ---
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

def agent_node(state: AgentState) -> AgentState:
    """核心推理节点：调用 LLM 决定下一步行动"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    """条件边：决定是继续执行工具还是结束"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "execute_tools"  # 有工具调用 → 执行工具
    return END                  # 无工具调用 → 结束

# --- 4. 构建图 ---
tool_node = ToolNode(tools)  # 自动处理工具调用

graph = StateGraph(AgentState)

# 添加节点
graph.add_node("agent", agent_node)
graph.add_node("execute_tools", tool_node)

# 设置入口
graph.set_entry_point("agent")

# 添加条件边（循环核心！）
graph.add_conditional_edges(
    "agent",
    should_continue,
    {
        "execute_tools": "execute_tools",
        END: END
    }
)

# 工具执行完毕后回到 Agent 继续推理
graph.add_edge("execute_tools", "agent")

# 编译
app = graph.compile()

# --- 5. 运行 ---
def run_agent(question: str):
    print(f"\n🔵 问题：{question}\n{'='*50}")
    final_state = app.invoke({
        "messages": [HumanMessage(content=question)]
    })
    answer = final_state["messages"][-1].content
    print(f"\n✅ 最终答案：{answer}")
    return answer

run_agent("用 Python 计算斐波那契数列前10项，并说明规律")

# ====================================================
# CrewAI 研究报告生成团队
# pip install crewai crewai-tools
# ====================================================
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
import os

os.environ["OPENAI_API_KEY"] = "your-key"
os.environ["SERPER_API_KEY"] = "your-serper-key"  # Google搜索API

# 工具
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# --- 1. 定义 Agent 角色 ---
researcher = Agent(
    role="资深研究员",
    goal="通过深度搜索收集关于{topic}的全面、准确的最新信息",
    backstory="""你是一位有15年经验的技术研究员，擅长从复杂信息中
    提炼关键洞察。你总是引用可靠来源，区分事实与观点。""",
    tools=[search_tool, web_tool],
    llm="gpt-4o",
    verbose=True,
    allow_delegation=False,   # 不允许再委派给其他 Agent
    max_iter=5,               # 最多迭代5次
)

analyst = Agent(
    role="战略分析师",
    goal="分析研究数据，识别趋势、机会和风险",
    backstory="""你是顶级咨询公司的战略分析师，擅长将数据转化为
    战略洞察，善用框架（SWOT、波特五力等）进行结构化分析。""",
    tools=[],
    llm="gpt-4o",
    verbose=True,
)

writer = Agent(
    role="技术写作专家",
    goal="将研究和分析转化为清晰、专业的报告",
    backstory="""你是技术博客作者和内容策略师，能将复杂技术内容
    转化为易读的专业报告，结构清晰，逻辑严密。""",
    tools=[],
    llm="gpt-4o",
    verbose=True,
)

# --- 2. 定义任务 ---
research_task = Task(
    description="""深入研究{topic}：
    1. 收集最新市场数据和技术进展
    2. 识别主要参与者和竞争格局  
    3. 整理3-5个关键发现，每个附带信息来源""",
    expected_output="包含关键数据和来源的结构化研究报告（Markdown格式）",
    agent=researcher,
    output_file="research.md",   # 自动保存到文件
)

analysis_task = Task(
    description="""基于研究员的发现，进行深度战略分析：
    1. SWOT 分析
    2. 未来12个月的核心趋势预测
    3. 对企业/开发者的实际建议""",
    expected_output="战略分析报告，包含 SWOT 矩阵和可执行建议",
    agent=analyst,
    context=[research_task],    # 依赖研究任务的输出作为上下文
)

writing_task = Task(
    description="""整合研究和分析内容，撰写一篇专业报告：
    - 标题和执行摘要（200字以内）
    - 研究发现（分章节）
    - 战略分析与建议
    - 结论与展望
    格式要求：Markdown，适合发布到技术博客""",
    expected_output="完整的专业分析报告，Markdown格式，2000字以上",
    agent=writer,
    context=[research_task, analysis_task],
    output_file="final_report.md",
)

# --- 3. 组建 Crew ---
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # 顺序执行：研究→分析→写作
    verbose=True,
    memory=True,                 # 开启跨任务记忆共享
    max_rpm=10,                  # 限速：每分钟最多10次API调用
)

# --- 4. 启动 ---
result = crew.kickoff(inputs={"topic": "2025年 AI Agent 商业化落地现状"})
print(result)

# ReAct 的 Prompt 模板（核心是格式约束）
REACT_TEMPLATE = """
你是一个智能助手，请按以下格式逐步解决问题：

Thought: 分析当前状态，思考下一步行动
Action: 选择工具名称（从 {tool_names} 中选择）
Action Input: 工具的输入参数
Observation: （工具返回的结果，自动填入）

重复以上步骤直到问题解决，最终输出：
Final Answer: 最终答案

可用工具：{tools}
问题：{input}
{agent_scratchpad}
"""

# ReAct 执行轨迹示例：
"""
Thought: 用户询问特斯拉股价，我需要先搜索当前价格
Action: web_search
Action Input: "Tesla TSLA stock price today"
Observation: TSLA 当前价格 $248.50，今日涨幅 +2.3%

Thought: 已获得股价，现在计算购买100股的费用
Action: calculator
Action Input: 248.50 * 100
Observation: 24850.0

Final Answer: 特斯拉(TSLA)当前股价为 $248.50，购买100股需要 $24,850。
"""

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 阶段一：Planner — 生成执行计划
PLANNER_PROMPT = """你是任务规划专家。将用户请求分解为具体的、可执行的步骤列表。
每步骤必须明确：做什么、用什么工具、期望什么输出。
以 JSON 数组格式返回，每项包含 step_id、action、tool、expected_output。"""

planner = ChatPromptTemplate.from_messages([
    ("system", PLANNER_PROMPT),
    ("human", "任务：{task}")
]) | llm

# 阶段二：Executor — 逐步执行
EXECUTOR_PROMPT = """根据计划执行当前步骤。
当前步骤：{current_step}
已完成步骤的结果：{past_results}
使用合适的工具完成任务。"""

# 阶段三：Replanner — 根据结果动态调整计划
REPLANNER_PROMPT = """根据执行结果检查计划是否需要调整。
原始计划：{original_plan}
已完成：{completed}
当前结果：{current_result}
如果任务已完成，返回 {{"done": true, "final_answer": "..."}}
如果需要调整，返回更新后的计划。"""

def run_plan_execute(task: str):
    # 1. 生成初始计划
    plan_response = planner.invoke({"task": task})
    print(f"📋 执行计划：{plan_response.content}")
    
    # 2. 逐步执行（此处简化，实际需循环执行每个步骤）
    # ... 循环执行逻辑 ...
    
    return plan_response.content

# Reflection Agent：执行 → 评估 → 改进的循环
REFLECTION_PROMPT = """你是一个严格的质量审查员。
评估以下输出是否满足要求，给出详细的改进建议。

原始任务：{task}
当前输出：{output}
评估标准：{criteria}

请返回：
1. 是否通过（pass/fail）
2. 问题列表（如果 fail）
3. 改进建议（如果 fail）"""

def reflection_agent(task: str, max_attempts: int = 3):
    llm = ChatOpenAI(model="gpt-4o")
    output = None
    
    for attempt in range(max_attempts):
        # 执行任务
        if attempt == 0:
            output = llm.invoke(f"完成以下任务：{task}").content
        else:
            output = llm.invoke(
                f"根据反馈改进：\n任务：{task}\n之前的输出：{output}\n改进建议：{feedback}"
            ).content
        
        # 自我反思
        reflection_chain = ChatPromptTemplate.from_messages([
            ("system", REFLECTION_PROMPT),
            ("human", "开始评估")
        ]) | llm
        
        review = reflection_chain.invoke({
            "task": task, "output": output,
            "criteria": "完整性、准确性、结构清晰度"
        })
        
        if "pass" in review.content.lower():
            print(f"✅ 第{attempt+1}次通过质量审查")
            return output
        else:
            feedback = review.content
            print(f"🔄 第{attempt+1}次反思，继续改进...")
    
    return output  # 达到最大次数，返回最佳结果

# ====================================================
# 🚀 生产级 Agent 原型模板
# 整合：LangGraph + Memory + Tools + Streaming
# ====================================================
from typing import Annotated, TypedDict, Optional
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_core.tools import tool
from langchain.memory import ConversationSummaryBufferMemory
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
import operator

# ── 1. State 定义 ─────────────────────────────────────
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    user_id: str                     # 用户标识（用于隔离记忆）
    task_context: Optional[dict]     # 任务上下文（规划信息等）
    iteration_count: int             # 迭代计数（防止无限循环）

# ── 2. 工具定义 ───────────────────────────────────────
@tool
def search_knowledge_base(query: str) -> str:
    """在内部知识库中搜索相关信息"""
    # 替换为真实的向量检索
    return f"[知识库] 关于'{query}'的相关内容..."

@tool
def execute_code(code: str, language: str = "python") -> str:
    """安全执行代码片段"""
    if language != "python":
        return "当前仅支持 Python"
    try:
        import io, contextlib
        buf = io.StringIO()
        with contextlib.redirect_stdout(buf):
            exec(code, {"__builtins__": {}})  # 限制内置函数
        return buf.getvalue() or "执行成功"
    except Exception as e:
        return f"错误: {e}"

@tool
def save_to_memory(key: str, value: str) -> str:
    """将重要信息保存到持久记忆"""
    # 替换为真实的存储
    return f"已保存：{key} = {value}"

tools = [search_knowledge_base, execute_code, save_to_memory]

# ── 3. LLM 配置 ────────────────────────────────────────
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,
    streaming=True,          # 开启流式输出
).bind_tools(tools)

SYSTEM_PROMPT = """你是一个专业、可靠的 AI 助手。

工作原则：
1. 遇到不确定的事实，使用搜索工具查证
2. 需要计算时，编写并执行代码
3. 重要信息及时保存到记忆
4. 每次行动前明确说明原因

限制：
- 每次对话最多进行10次工具调用
- 不执行任何可能造成损害的操作"""

# ── 4. 节点函数 ────────────────────────────────────────
def agent_node(state: AgentState) -> AgentState:
    """推理节点：带系统提示的 LLM 调用"""
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    
    # 安全检查：防止无限循环
    if state.get("iteration_count", 0) >= 10:
        from langchain_core.messages import AIMessage
        return {"messages": [AIMessage(content="已达到最大迭代次数，请重新提问。")],
                "iteration_count": state.get("iteration_count", 0) + 1}
    
    response = llm.invoke(messages)
    return {
        "messages": [response],
        "iteration_count": state.get("iteration_count", 0) + 1
    }

def route(state: AgentState) -> str:
    """路由函数：决定下一步"""
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

# ── 5. 构建图 ──────────────────────────────────────────
tool_node = ToolNode(tools)
checkpointer = MemorySaver()  # 持久化状态（生产可换 PostgresSaver）

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", route, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")

# compile 时注入 checkpointer，实现多轮对话记忆
app = builder.compile(checkpointer=checkpointer)

# ── 6. 多轮对话接口 ────────────────────────────────────
def chat(user_id: str, message: str):
    """支持多轮对话的入口函数"""
    config = {"configurable": {"thread_id": user_id}}  # 用 user_id 隔离会话
    
    result = app.invoke(
        {
            "messages": [HumanMessage(content=message)],
            "user_id": user_id,
            "task_context": {},
            "iteration_count": 0,
        },
        config=config
    )
    return result["messages"][-1].content

# ── 7. 使用示例 ────────────────────────────────────────
if __name__ == "__main__":
    user = "user_001"
    
    # 第一轮
    print(chat(user, "帮我搜索一下 LangGraph 的最新文档"))
    
    # 第二轮（自动携带上下文）
    print(chat(user, "根据你刚才找到的内容，写一个简单示例"))
    
    # 第三轮
    print(chat(user, "把这个示例保存到记忆里"))