3.Agent开发基础

from langchain.agents import create_agent
from langchain_core.tools import tool

# ---------- 1. ReAct（create_agent 默认就是 ReAct）----------
# Agent 边推理边调工具，根据工具结果决定下一步

@tool
def search_weather(city: str) -> str:
    """查询城市天气"""
    return f"{city}：晴，25°C"

react_agent = create_agent(
    model=model_openai,
    tools=[search_weather],
    system_prompt="你是天气助手，根据需要调用工具查询天气。",
    max_iterations=5,  # 防死循环
)
result = react_agent.invoke({
    "messages": [{"role": "user", "content": "北京天气怎么样？"}]
})


# ---------- 2. CoT 思维链（不调工具，靠模型推理）----------
# 用带推理能力的模型 + Prompt 引导，不走 Agent 循环

from langchain_core.prompts import ChatPromptTemplate

cot_prompt = ChatPromptTemplate.from_messages([
    ("system", "一步步推理后再给答案。"),
    ("user", "{question}"),
])
cot_chain = cot_prompt | model_openai
result = cot_chain.invoke({"question": "一个水池有两个进水管..."})  # 数学推理


# ---------- 3. ToT 思维树（无内置，需自己组装）----------
# 生成多个候选方案 → 评分 → 选最优

@tool
def generate_options(problem: str) -> list[str]:
    """生成多个候选方案"""
    # 实际是多次调用模型
    return ["方案A", "方案B", "方案C"]

@tool
def evaluate_option(option: str) -> float:
    """对方案评分"""
    return 0.85

# 自行循环：generate → evaluate → 取最高分
# LangChain 无内置 ToT Agent


# ---------- 4. Reflection 反思（用中间件在 after 钩子批评-修订）----------
# 自定义中间件，agent 执行后让 Reflector 批评，Generator 修订

from langchain.agents.middleware import AgentMiddleware

class ReflectionMiddleware(AgentMiddleware):
    async def after_agent(self, state, runtime):
        output = state["messages"][-1].content
        # 让模型批评自己的输出
        critique = await model_openai.ainvoke(
            f"批评这段回答的不足：{output}"
        )
        # 根据批评修订
        revised = await model_openai.ainvoke(
            f"根据批评修订：{output}\n批评：{critique.content}"
        )
        state["messages"][-1].content = revised.content
        return state

reflect_agent = create_agent(
    model=model_openai,
    tools=[],
    system_prompt="你是写作助手。",
    middleware=[ReflectionMiddleware()],
)


# ---------- 5. Plan-and-Execute（无内置完整版，自己组装三部分）----------
# Planner 规划 → Executor 逐步执行 → Replanner 动态纠偏

@tool
def make_plan(task: str) -> list[str]:
    """生成步骤清单"""
    return ["步骤1: 收集资料", "步骤2: 分析", "步骤3: 生成报告"]

@tool
def execute_step(step: str) -> str:
    """执行单步（内部可跑一个 ReAct agent）"""
    return f"{step} 已完成"

plan = make_plan.invoke({"task": "写市场分析报告"})
for step in plan:
    result = execute_step.invoke({"step": step})
    # Replanner：根据 result 决定继续、跳过或改 plan
    # 实际用一次 LLM 调用判断


# ---------- 6. Function Calling（绑工具直接调，不走 Agent 循环）----------
# 模型原生 function calling，单步任务

bound_model = model_openai.bind_tools([search_weather])
response = bound_model.invoke("北京天气怎么样？")
# response.tool_calls 直接拿到结构化参数，单次调用

from openai import OpenAI
client = OpenAI(api_key="你的key",
                base_url="https://platform.aitools.cfd/api/v1")

response = client.chat.completions.create(
    model="zhipu/glm-4-flash",
    messages=[
      {"role": "system", "content": "你是张三的AI助手"},
      {"role": "user", "content": "你好，你是谁？"},
    ],
    stream=False
)
print(response.choices[0].message.content)

from openai import OpenAI

client = OpenAI(api_key="你的API密钥")

def simple_agent(user_input: str) -> str:
    """最简单的 Agent：接收输入 → LLM 思考 → 返回输出"""

    system_prompt = """你是一个有帮助的AI助手。
    回答问题时，先分析用户需求，再给出清晰的结构化回答。"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input}
        ]
    )

    return response.choices[0].message.content

# 测试
result = simple_agent("我想学习Python，应该从哪里开始？")
print(result)

from openai import OpenAI
client = OpenAI(api_key="你的API密钥")

class MemoryAgent:
  def __init__(self):
    # 系统内存保存上下文
    self.messages = [
      {"role": "system", "content": "你是一个有帮助的AI助手，能记住之前的对话内容。"}
    ]

  def chat(self, user_input: str) -> str:
    # 感知：记录用户输入
    self.messages.append({"role": "user", "content": user_input})

    # 思考：LLM 推理（自动包含历史记忆）
    response = client.chat.completions.create(
        model="zhipu/glm-4-flash",
        messages=self.messages
    )

    assistant_message = response.choices[0].message.content

    # 记忆：保存助手回复
    self.messages.append({"role": "assistant", "content": assistant_message})

    return assistant_message

  def show_memory(self):
    """查看 Agent 的记忆"""
    print(f"记忆中有 {len(self.messages)} 条消息")
    for msg in self.messages[1:]:  # 跳过 system prompt
      role = "👤用户" if msg["role"] == "user" else "🤖助手"
      print(f"{role}: {msg['content'][:50]}...")

# 测试
agent = MemoryAgent()
print(agent.chat("我叫小明，我在学Python"))
print(agent.chat("我叫什么名字？"))  # Agent 能记住
agent.show_memory()

import json
from openai import OpenAI

client = OpenAI(api_key="你的key",
                base_url="https://platform.aitools.cfd/api/v1")

# 定义工具函数
def get_weather(city: str) -> str:
  """模拟天气查询（实际项目中调用真实API）"""
  weather_data = {
    "北京": "晴天，25°C",
    "上海": "多云，22°C",
    "深圳": "小雨，28°C"
  }
  return weather_data.get(city, f"{city}：暂无天气数据")

def calculate(expression: str) -> str:
  """安全计算数学表达式"""
  try:
    allowed = set("0123456789+-*/.() ")
    if all(c in allowed for c in expression):
      return str(eval(expression))
    return "不安全的表达式"
  except:
    return "计算错误"

# 工具注册表
tools = [
  {
    "type": "function",
    "function": {
      "name": "get_weather",
      "description": "查询指定城市的天气",
      "parameters": {
        "type": "object",
        "properties": {
          "city": {"type": "string", "description": "城市名称"}
        },
        "required": ["city"]
      }
    }
  },
  {
    "type": "function",
    "function": {
      "name": "calculate",
      "description": "计算数学表达式",
      "parameters": {
        "type": "object",
        "properties": {
          "expression": {"type": "string", "description": "数学表达式"}
        },
        "required": ["expression"]
      }
    }
  }
]

# 工具执行映射
tool_map = {
  "get_weather": get_weather,
  "calculate": calculate
}

class ToolAgent:
  def __init__(self):
    self.messages = [
      {"role": "system", "content": "你是一个能查询天气和做计算的AI助手。根据用户需求选择合适的工具。"}
    ]

  def chat(self, user_input: str) -> str:
    self.messages.append({"role": "user", "content": user_input})

    # 第一步：LLM 决定是否需要调用工具
    response = client.chat.completions.create(
        model="openai/gpt-oss-20b",
        messages=self.messages,
        tools=tools
    )

    message = response.choices[0].message

    # 第二步：如果 LLM 决定调用工具
    if message.tool_calls:
      for tool_call in message.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)

        print(f"  🔧 调用工具: {func_name}({func_args})")

        # 执行工具：展开字典为关键字参数
        result = tool_map[func_name](**func_args)
        print(f"  📋 工具结果: {result}")

        # 把工具结果告诉 LLM
        self.messages.append(message)
        self.messages.append({
          "role": "tool",
          "tool_call_id": tool_call.id,
          "content": result
        })

      # 第三步：LLM 基于工具结果生成最终回答
      final_response = client.chat.completions.create(
          model="openai/gpt-oss-20b",
          messages=self.messages
      )
      final_message = final_response.choices[0].message.content
      self.messages.append({"role": "assistant", "content": final_message})
      return final_message

    # 不需要工具，直接回答
    self.messages.append({"role": "assistant", "content": message.content})
    return message.content

# 测试
agent = ToolAgent()
print(agent.chat("北京今天天气怎么样？"))
print("---")
print(agent.chat("帮我算一下 (123 + 456) * 2"))

from openai import OpenAI

client = OpenAI(api_key="你的key",
                base_url="https://platform.aitools.cfd/api/v1")

class PlanningAgent:
  """能自主规划多步骤任务的 Agent"""

  def __init__(self):
    self.messages = [
      {"role": "system", "content": """你是一个智能规划助手。
当收到复杂任务时，你会：
1. 先制定执行计划（列出步骤）
2. 逐步执行每个步骤
3. 每步完成后评估结果
4. 如有偏差，调整计划

请用以下格式输出计划：
📋 计划：
1. [步骤1]
2. [步骤2]
...

然后逐步执行。"""}
    ]
    self.max_iterations = 5  # 防止无限循环

  def run(self, task: str) -> str:
    self.messages.append({"role": "user", "content": task})

    for i in range(self.max_iterations):
      response = client.chat.completions.create(
          model="openai/gpt-oss-20b",
          messages=self.messages
      )

      reply = response.choices[0].message.content
      self.messages.append({"role": "assistant", "content": reply})

      print(f"\n--- 迭代 {i+1} ---")
      print(reply)

      # 检查是否完成
      check = client.chat.completions.create(
          model="openai/gpt-oss-20b",
          messages=self.messages + [
            {"role": "user", "content": "任务是否已完成？只回答'是'或'否'"}
          ]
      )

      if check.choices[0].message.content.strip().startswith("是"):
        print("\n✅ 任务完成！")
        return reply

      # 继续执行
      self.messages.append({"role": "user", "content": "请继续执行下一步。"})

    return "⚠️ 达到最大迭代次数"

# 测试
agent = PlanningAgent()
agent.run("帮我分析学习Python的完整路线图，包括每个阶段的学习重点和时间安排")

# 安装：pip install langchain langchain-openai

from langchain.agents import create_agent
from langchain_core.tools import tool

from _5_机器学习._04_langchain.common import model_openai


# 第一步：定义工具
@tool
def search_movie(title: str) -> str:
  """搜索电影信息"""
  movies = {
    "星际穿越": "2014年，导演：克里斯托弗·诺兰，评分：9.4",
    "盗梦空间": "2010年，导演：克里斯托弗·诺兰，评分：9.3"
  }
  return movies.get(title, "未找到该电影")

@tool
def recommend_movie(genre: str) -> str:
  """根据类型推荐电影"""
  recs = {
    "科幻": "推荐：《星际穿越》《银翼杀手2049》《降临》",
    "喜剧": "推荐：《布达佩斯大饭店》《功夫》《让子弹飞》"
  }
  return recs.get(genre, "暂无该类型推荐")

# 第二步：创建 Agent
tools = [search_movie, recommend_movie]
agent = create_agent(
    model=model_openai,
    tools=tools,
    system_prompt="你是一个电影推荐助手，能搜索电影信息和根据类型推荐电影。"
)

# 第三步：运行
result = agent.invoke({
  "messages": [{"role": "user", "content": "我想看一部科幻电影，帮我搜索星际穿越的信息"}]
})

for msg in result["messages"]:
  if hasattr(msg, "content") and msg.content:
    print(f"[{msg.type}] {msg.content}")

# 技巧1：开启 verbose 模式查看 Agent 思考过程
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 技巧2：打印完整消息链
for msg in agent.messages:
    print(f"[{msg['role']}] {msg['content'][:100]}")

# 技巧3：限制迭代次数防止死循环
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,  # 最多5轮
    early_stopping_method="generate"  # 超限后生成最终回答
)

# 技巧4：添加重试机制
import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def call_llm_with_retry(messages):
    return client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages
    )