Agent基础与架构¶

# Agent应用场景示例

agent_applications = {
    "代码助手": "理解需求 → 搜索文档 → 编写代码 → 运行测试 → 修复bug",
    "研究助手": "接收课题 → 搜索文献 → 阅读论文 → 提取要点 → 生成报告",
    "数据分析": "理解问题 → 查询数据库 → 分析数据 → 生成图表 → 撰写结论",
    "客服系统": "理解问题 → 查询知识库 → 检索订单 → 生成回复 → 跟进满意度",
    "自动化运维": "监控告警 → 分析日志 → 定位问题 → 执行修复 → 验证结果",
}

"""
ReAct范式的基础实现
展示Thought-Action-Observation循环
"""

from openai import OpenAI

client = OpenAI()

REACT_SYSTEM_PROMPT = """你是一个有用的AI助手，你使用ReAct方法来回答问题。

对于每个步骤，你必须严格按照以下格式输出：

Thought: <你的推理过程>
Action: <工具名称>(<参数>)

当你已经有足够信息回答问题时，使用：
Thought: <你的推理过程>
Action: finish(<最终答案>)

可用工具：
- search(query): 搜索信息
- calculate(expression): 计算数学表达式
- get_weather(city): 获取城市天气

重要：每次只输出一个Thought和一个Action。"""

def parse_action(response_text: str) -> tuple[str, str]:
    """解析LLM输出中的Action"""
    lines = response_text.strip().split("\n")
    for line in lines:
        if line.startswith("Action:"):
            action_str = line[len("Action:"):].strip()
            # 解析函数名和参数，容错处理格式异常
            if "(" not in action_str or ")" not in action_str:
                return action_str, ""  # 无参数格式，返回整行作为函数名
            paren_idx = action_str.index("(")
            func_name = action_str[:paren_idx].strip()
            args = action_str[paren_idx + 1:-1].strip()
            return func_name, args
    return "", ""

def execute_tool(func_name: str, args: str) -> str:
    """执行工具调用（模拟实现）"""
    import ast
    import operator
    def _safe_calc(expr: str) -> str:
        """基于AST的安全数学表达式计算（仅允许数字和四则运算）。
        不使用eval，通过遍历AST节点确保安全性。"""
        # 定义允许的运算符（仅四则运算）
        allowed_ops = {
            ast.Add: operator.add, ast.Sub: operator.sub,
            ast.Mult: operator.mul, ast.Div: operator.truediv,
        }
        def _eval_node(node):
            """递归求值AST节点"""
            if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
                return node.value
            elif isinstance(node, ast.BinOp) and type(node.op) in allowed_ops:
                return allowed_ops[type(node.op)](_eval_node(node.left), _eval_node(node.right))
            elif isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
                return -_eval_node(node.operand)
            else:
                raise ValueError(f"不支持的表达式: {expr}")
        try:
            tree = ast.parse(expr.strip(), mode='eval')
            result = _eval_node(tree.body)
        except (SyntaxError, ValueError) as e:
            return f"计算错误: {e}"
        return str(result)
    # lambda是匿名函数：lambda 参数: 返回值，这里用作字典值实现工具分发
    tools = {
        "search": lambda q: f"搜索结果: {q}相关的信息...",
        "calculate": lambda expr: f"计算结果: {_safe_calc(expr)}",
        "get_weather": lambda city: f"{city}今天晴，温度15-25°C，湿度40%",
        "finish": lambda answer: answer,
    }
    if func_name in tools:
        return tools[func_name](args)
    return f"未知工具: {func_name}"

def react_agent(question: str, max_steps: int = 5) -> str:
    """ReAct Agent主循环"""
    messages = [
        {"role": "system", "content": REACT_SYSTEM_PROMPT},
        {"role": "user", "content": question},
    ]

    for step in range(max_steps):
        print(f"\n--- Step {step + 1} ---")

        # 1. 让LLM推理并选择动作
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0,
        )
        assistant_msg = response.choices[0].message.content
        print(assistant_msg)

        # 2. 解析动作
        func_name, args = parse_action(assistant_msg)

        # 3. 如果是finish动作，返回结果
        if func_name == "finish":
            return args

        # 4. 执行工具并获取观察结果
        observation = execute_tool(func_name, args)
        print(f"Observation: {observation}")

        # 5. 将结果加入对话历史
        messages.append({"role": "assistant", "content": assistant_msg})
        messages.append({"role": "user", "content": f"Observation: {observation}"})

    return "达到最大步骤数，未能完成任务"

# 运行示例
if __name__ == "__main__":
    result = react_agent("北京今天天气怎么样？如果温度超过20度，计算20*1.8+32等于多少华氏度")
    print(f"\n最终结果: {result}")

"""
Agent的LLM组件 - 推理引擎
"""

from openai import OpenAI
from typing import Any

class AgentLLM:
    """Agent的LLM推理引擎"""

    def __init__(self, model: str = "gpt-4o", temperature: float = 0):
        self.client = OpenAI()
        self.model = model
        self.temperature = temperature

    def think(self, messages: list[dict], tools: list[dict] | None = None) -> Any:
        """执行一次推理"""
        kwargs = {
            "model": self.model,
            "messages": messages,
            "temperature": self.temperature,
        }
        if tools:
            kwargs["tools"] = tools
            kwargs["tool_choice"] = "auto"

        # **kwargs将字典解包为关键字参数，等价于create(model=..., messages=..., ...)
        response = self.client.chat.completions.create(**kwargs)
        return response.choices[0].message

    def stream_think(self, messages: list[dict]):
        """流式推理（用于实时输出）"""
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=self.temperature,
            stream=True,
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

"""
Agent工具系统
"""

import json
from typing import Callable
from dataclasses import dataclass, field

@dataclass
class Tool:
    """工具定义"""
    name: str
    description: str
    parameters: dict  # JSON Schema格式
    function: Callable  # 实际执行的函数

class ToolRegistry:
    """工具注册中心"""

    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, name: str, description: str, parameters: dict):
        """装饰器：注册工具"""
        def decorator(func: Callable):
            self._tools[name] = Tool(
                name=name,
                description=description,
                parameters=parameters,
                function=func,
            )
            return func
        return decorator

    def execute(self, name: str, arguments: dict) -> str:
        """执行指定工具"""
        if name not in self._tools:
            return json.dumps({"error": f"工具 {name} 不存在"})
        try:
            result = self._tools[name].function(**arguments)
            # 三元表达式: A if 条件 else B；isinstance第二个参数用元组表示"是其中任一类型"
            return json.dumps(result, ensure_ascii=False) if isinstance(result, (dict, list)) else str(result)
        except Exception as e:
            return json.dumps({"error": f"工具执行失败: {str(e)}"})

    def get_openai_tools(self) -> list[dict]:
        """导出为OpenAI Tool Calling格式"""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters,
                },
            }
            for tool in self._tools.values()
        ]

# === 使用示例 ===

registry = ToolRegistry()

@registry.register(
    name="search_web",
    description="搜索互联网信息",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词",
            }
        },
        "required": ["query"],
    },
)
def search_web(query: str) -> dict:
    """模拟网络搜索"""
    return {"results": [f"搜索 '{query}' 的结果1", f"搜索 '{query}' 的结果2"]}

@registry.register(
    name="calculate",
    description="执行数学计算",
    parameters={
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "数学表达式，如 '2+3*4'",
            }
        },
        "required": ["expression"],
    },
)
def calculate(expression: str) -> dict:
    """安全地计算数学表达式

    ⚠️ 安全提示: 即使做了字符白名单，eval() 仍存在安全隐患。
    生产环境请使用 simpleeval 库:
        from simpleeval import simple_eval
        return {"result": simple_eval(expression)}
    """
    allowed_chars = set("0123456789+-*/.() ")
    # all(生成器): 检查expression每个字符是否都在白名单中，有一个不在就返回False
    if not all(c in allowed_chars for c in expression):
        return {"error": "不支持的字符"}
    # 限制表达式长度，降低风险
    if len(expression) > 100:
        return {"error": "表达式过长"}
    return {"result": eval(expression)}  # noqa: S307 — 生产环境请替换为 simpleeval

# 查看生成的OpenAI格式工具定义
print(json.dumps(registry.get_openai_tools(), indent=2, ensure_ascii=False))

"""
Agent记忆系统
支持短期记忆（对话历史）和长期记忆（向量存储）
"""

from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class MemoryItem:
    """记忆条目"""
    content: str
    role: str  # user/assistant/system/tool
    # field(default_factory=函数): 每次创建实例时调用该函数生成默认值
    # 不能直接写 timestamp: datetime = datetime.now()，否则所有实例共享同一个时间
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: dict = field(default_factory=dict)  # 每个实例独立的空字典，不会共享引用

class ShortTermMemory:
    """短期记忆 - 基于滑动窗口的对话历史"""

    def __init__(self, max_messages: int = 20):
        self.messages: list[MemoryItem] = []
        self.max_messages = max_messages

    def add(self, role: str, content: str, **metadata):
        """添加记忆"""
        self.messages.append(MemoryItem(
            content=content, role=role, metadata=metadata
        ))
        # 超出限制时，保留system消息，裁剪早期消息
        if len(self.messages) > self.max_messages:
            # 列表推导式: [表达式 for 变量 in 可迭代对象 if 条件]，从列表中按条件筛选
            system_msgs = [m for m in self.messages if m.role == "system"]
            other_msgs = [m for m in self.messages if m.role != "system"]
            # 计算非system消息的保留数量，确保不为负数
            keep_count = max(0, self.max_messages - len(system_msgs))
            if keep_count > 0:
                # 负数切片 [-n:] 表示取列表最后n个元素
                self.messages = system_msgs + other_msgs[-keep_count:]
            else:
                # 极端情况：system消息已超过上限，只保留最近的system消息
                self.messages = system_msgs[-self.max_messages:]

    def get_messages(self) -> list[dict]:
        """获取OpenAI格式的消息列表"""
        # 列表推导: 将MemoryItem对象列表转换为字典列表（OpenAI API要求的格式）
        return [{"role": m.role, "content": m.content} for m in self.messages]

    def clear(self):
        """清空记忆"""
        self.messages.clear()

    def summarize(self, llm) -> str:
        """使用LLM对历史对话进行摘要压缩"""
        # "分隔符".join(列表): 用分隔符把列表元素拼接成一个字符串
        history = "\n".join([f"{m.role}: {m.content}" for m in self.messages])
        summary_prompt = f"请将以下对话历史压缩为简洁的摘要：\n{history}"
        # 调用LLM生成摘要
        return llm.think([{"role": "user", "content": summary_prompt}]).content

class LongTermMemory:
    """长期记忆 - 基于向量检索的持久化记忆"""

    def __init__(self):
        # 简化实现，实际应使用向量数据库（Chroma/Milvus等）
        self.memories: list[dict] = []

    def store(self, content: str, metadata: dict | None = None):
        """存储长期记忆"""
        self.memories.append({
            "content": content,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat(),
        })

    def search(self, query: str, top_k: int = 5) -> list[dict]:
        """搜索相关记忆（简化版，实际应用向量相似度）"""
        # 简单的关键词匹配，实际应使用embedding + cosine similarity
        results = []
        for memory in self.memories:
            # any(生成器): 只要query中有任意一个词出现在记忆内容里，就返回True
            if any(word in memory["content"] for word in query.split()):
                results.append(memory)
        return results[:top_k]

    def save_to_file(self, filepath: str):
        """持久化存储到文件"""
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(self.memories, f, ensure_ascii=False, indent=2)

    def load_from_file(self, filepath: str):
        """从文件加载"""
        with open(filepath, "r", encoding="utf-8") as f:
            self.memories = json.load(f)

# 使用示例
memory = ShortTermMemory(max_messages=10)
memory.add("system", "你是一个有用的AI助手。")
memory.add("user", "你好，请帮我查一下Python的最新版本。")
memory.add("assistant", "Python最新版本请以官方发布为准，建议查看python.org获取最新信息。")

print("对话历史:")
for msg in memory.get_messages():
    print(f"  {msg['role']}: {msg['content']}")

"""
Agent规划系统
支持任务分解和执行计划生成

⚠️ 关键设计：规划时必须将可用工具列表注入 Prompt，
否则 LLM 会幻觉出不存在的工具名，导致计划无法执行。
这与 LangGraph plan-and-execute、AutoGPT Commands 注入、
CrewAI tools 绑定等业界实践一致。
"""

import json
from openai import OpenAI

client = OpenAI()

PLANNING_PROMPT = """你是一个任务规划专家。请将用户的复杂任务分解为具体的执行步骤。

要求：
1. 每个步骤应该是原子化的、可执行的
2. 步骤之间的依赖关系要清晰
3. 每个步骤的 "tool" 字段只能从下方【可用工具】列表中选择
4. 输出JSON格式的计划

【可用工具】
{available_tools}

输出格式：
{{
    "goal": "最终目标",
    "steps": [
        {{
            "id": 1,
            "action": "具体动作描述",
            "tool": "需要使用的工具（必须从可用工具中选择）",
            "depends_on": [],
            "expected_output": "期望输出"
        }}
    ]
}}"""

def create_plan(task: str, tools: list[dict] | None = None) -> dict:
    """使用LLM创建执行计划

    Args:
        task: 要规划的任务描述
        tools: 可用工具列表，每个工具包含 name 和 description。
               规划器只会使用此列表中的工具，避免幻觉出不存在的工具。
    """
    # 格式化工具描述，注入到 Prompt 中
    if tools:
        # f-string中可以用['key']访问字典值；.join()拼接生成器产出的每一行
        tools_text = "\n".join(
            f"- {t['name']}: {t['description']}" for t in tools
        )
    else:
        tools_text = "（无特定工具约束，所有步骤由LLM直接推理完成）"

    prompt = PLANNING_PROMPT.format(available_tools=tools_text)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": prompt},
            {"role": "user", "content": f"请为以下任务制定执行计划：\n{task}"},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    return json.loads(response.choices[0].message.content)

# === 示例：定义可用工具并生成计划 ===

available_tools = [
    {"name": "web_search", "description": "通过关键词搜索互联网信息"},
    {"name": "sql_query", "description": "执行SQL查询获取结构化数据"},
    {"name": "calculator", "description": "执行数学计算和财务指标分析"},
    {"name": "document_writer", "description": "生成格式化的文档和报告"},
]

plan = create_plan(
    "帮我分析某公司2024年的财务数据，并生成一份投资建议报告",
    tools=available_tools,
)
print(json.dumps(plan, indent=2, ensure_ascii=False))

# 预期输出类似：
# {
#   "goal": "分析公司财务数据并生成投资建议",
#   "steps": [
#     {"id": 1, "action": "搜索公司基本信息", "tool": "web_search", ...},
#     {"id": 2, "action": "查询财务报表数据", "tool": "sql_query", ...},
#     {"id": 3, "action": "计算关键财务指标", "tool": "calculator", ...},
#     {"id": 4, "action": "生成投资建议报告", "tool": "document_writer", ...}
#   ]
# }

"""
路由模式：根据输入意图分发到不同的处理逻辑
"""

from openai import OpenAI
import json

client = OpenAI()

def route_request(user_input: str) -> str:
    """路由Agent - 根据输入意图分发处理"""

    # 使用LLM判断意图
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """分析用户输入的意图，返回JSON格式:
{"intent": "code|search|math|chat", "confidence": 0.0-1.0, "reason": "判断原因"}"""},
            {"role": "user", "content": user_input},
        ],
        response_format={"type": "json_object"},
    )

    intent_data = json.loads(response.choices[0].message.content)
    intent = intent_data["intent"]

    # 根据意图路由到不同处理器
    handlers = {
        "code": handle_code_request,
        "search": handle_search_request,
        "math": handle_math_request,
        "chat": handle_chat_request,
    }

    # 第二个参数是默认值：当LLM返回了意料之外的intent时，降级为闲聊处理，避免KeyError
    handler = handlers.get(intent, handle_chat_request)
    return handler(user_input)

def handle_code_request(query: str) -> str:
    """处理编程相关请求"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个Python编程专家，提供高质量的代码解决方案。"},
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content

def handle_search_request(query: str) -> str:
    return f"[搜索模块] 正在搜索: {query}"

def handle_math_request(query: str) -> str:
    return f"[数学模块] 正在计算: {query}"

def handle_chat_request(query: str) -> str:
    return f"[对话模块] 正在处理: {query}"

"""
反思模式：Agent生成 → 自我审查 → 改进
适用于代码生成、文章写作等需要高质量输出的场景
"""

from openai import OpenAI

client = OpenAI()

def reflect_and_improve(task: str, max_iterations: int = 3) -> str:
    """反思改进循环"""

    # 第1步：生成初始输出
    generation_response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个专业的Python开发者。请根据任务要求编写代码。"},
            {"role": "user", "content": task},
        ],
    )
    current_output = generation_response.choices[0].message.content
    print(f"=== 初始生成 ===\n{current_output}\n")

    for i in range(max_iterations):
        # 第2步：自我审查
        review_response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": """你是一个严格的代码审查专家。请审查以下代码，指出问题并给出具体改进建议。
如果代码已经足够好，回复"APPROVED"。
否则返回具体的问题列表和改进建议。"""},
                {"role": "user", "content": f"任务: {task}\n\n代码:\n{current_output}"},
            ],
        )
        review = review_response.choices[0].message.content
        print(f"=== 审查 {i + 1} ===\n{review}\n")

        # 如果审查通过，退出循环
        if "APPROVED" in review:
            print("✅ 审查通过!")
            break

        # 第3步：根据审查意见改进
        improve_response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "请根据审查意见改进代码。"},
                {"role": "user", "content": f"原始任务: {task}\n\n当前代码:\n{current_output}\n\n审查意见:\n{review}"},
            ],
        )
        current_output = improve_response.choices[0].message.content
        print(f"=== 改进 {i + 1} ===\n{current_output}\n")

    return current_output

# 使用示例
result = reflect_and_improve("编写一个线程安全的LRU缓存类")

"""
工具使用模式：基于OpenAI Tool Calling
这是最标准的Agent工具调用方式
"""

from openai import OpenAI
import json

client = OpenAI()

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "获取指定股票的当前价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "股票代码，如 AAPL, GOOGL",
                    },
                    "market": {
                        "type": "string",
                        "enum": ["US", "CN", "HK"],
                        "description": "市场，US=美股, CN=A股, HK=港股",
                    },
                },
                "required": ["symbol"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "analyze_sentiment",
            "description": "分析文本的情感倾向",
            "parameters": {
                "type": "object",
                "properties": {
                    "text": {
                        "type": "string",
                        "description": "要分析的文本",
                    },
                },
                "required": ["text"],
            },
        },
    },
]

def get_stock_price(symbol: str, market: str = "US") -> dict:
    """模拟获取股票价格"""
    prices = {"AAPL": 198.5, "GOOGL": 175.2, "600519": 1680.0}
    return {"symbol": symbol, "price": prices.get(symbol, 0), "market": market, "currency": "USD" if market == "US" else "CNY"}

def analyze_sentiment(text: str) -> dict:
    """模拟情感分析"""
    return {"text": text[:50], "sentiment": "positive", "score": 0.85}

# 工具函数映射
tool_functions = {
    "get_stock_price": get_stock_price,
    "analyze_sentiment": analyze_sentiment,
}

def tool_use_agent(user_message: str) -> str:
    """工具使用Agent"""
    messages = [
        {"role": "system", "content": "你是一个金融助手，可以查询股票价格和分析市场情感。"},
        {"role": "user", "content": user_message},
    ]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
        )

        message = response.choices[0].message

        # 如果没有工具调用，返回最终结果
        if not message.tool_calls:
            return message.content

        # 处理所有工具调用（支持并行调用）
        messages.append(message)

        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_functions[func_name](**func_args)

            # 将结果返回给LLM
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })

# 运行
answer = tool_use_agent("帮我查一下苹果股票现在多少钱？")
print(f"回答: {answer}")

"""
提示链模式：串行LLM调用管道
适用于流程固定、每步需要高质量输出的任务（如内容创作、数据处理）

参考：Anthropic "Building Effective Agents" (2025) — Prompt Chaining 模式
"""

from openai import OpenAI
import json

client = OpenAI()

def prompt_chain(topic: str) -> dict:
    """
    内容创作提示链：大纲 → 草稿 → 审校 → 终稿

    每一步专注于单一目标，通过"门控检查"决定是否进入下一步。
    """

    # Step 1: 生成大纲
    outline_resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个内容策划专家。请为给定主题生成详细的文章大纲。"},
            {"role": "user", "content": f"请为以下主题生成大纲，包含3-5个一级标题和各自的要点：\n{topic}"},
        ],
        temperature=0.7,
    )
    outline = outline_resp.choices[0].message.content
    print(f"✅ Step 1 - 大纲生成完成")

    # 门控检查：大纲质量验证
    gate_resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": '判断大纲是否结构完整、逻辑清晰。返回JSON：{"pass": true/false, "reason": "..."}'},
            {"role": "user", "content": outline},
        ],
        response_format={"type": "json_object"},
    )
    gate = json.loads(gate_resp.choices[0].message.content)
    if not gate.get("pass", False):
        print(f"⚠️ 大纲未通过门控检查: {gate.get('reason')}")
        # 可选：重新生成或人工修正
        # 这里简化处理，继续执行

    # Step 2: 基于大纲撰写草稿
    draft_resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个专业技术写手。请基于大纲撰写完整文章。"},
            {"role": "user", "content": f"大纲：\n{outline}\n\n请据此撰写完整文章，每个章节200-300字。"},
        ],
        temperature=0.7,
    )
    draft = draft_resp.choices[0].message.content
    print(f"✅ Step 2 - 草稿撰写完成")

    # Step 3: 审校与优化
    review_resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个资深编辑。请审校文章，修正事实错误、优化表达、统一风格。直接输出修改后的完整文章。"},
            {"role": "user", "content": draft},
        ],
        temperature=0.3,
    )
    final = review_resp.choices[0].message.content
    print(f"✅ Step 3 - 审校优化完成")

    return {"outline": outline, "draft": draft, "final": final}

# 使用示例
result = prompt_chain("AI Agent在企业客服中的应用")
print(f"\n📋 终稿:\n{result['final'][:500]}...")

"""
并行化模式：多个LLM调用并行执行
子模式1 - 分段并行：不同子任务同时执行
子模式2 - 投票并行：同一任务多次执行取最优

参考：Anthropic "Building Effective Agents" (2025) — Parallelization 模式
"""

import asyncio  # Python标准异步库，提供事件循环和async/await协程支持
from openai import AsyncOpenAI  # OpenAI异步客户端，返回协程对象，需配合await使用

aclient = AsyncOpenAI()

# === 子模式1：分段并行 ===

# async def定义协程函数，必须在事件循环中通过await调用或用asyncio.run执行
async def analyze_parallel(text: str) -> dict:
    """
    对同一段文本，并行执行多个独立分析任务
    每个分析维度互不依赖 → 并行处理
    """

    async def _call(system_prompt: str, user_prompt: str) -> str:
        response = await aclient.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0,
        )
        return response.choices[0].message.content

    # 三个独立分析任务并行执行
    sentiment_task = _call("你是情感分析专家，判断文本的情感倾向（正面/负面/中性）并给出理由。", text)
    keyword_task = _call("你是关键词提取专家，提取文本的5个核心关键词及每个关键词的重要性说明。", text)
    summary_task = _call("你是摘要专家，用3句话概括文本核心内容。", text)

    # asyncio.gather 并发等待所有任务完成
    sentiment, keywords, summary = await asyncio.gather(
        sentiment_task, keyword_task, summary_task
    )

    return {"sentiment": sentiment, "keywords": keywords, "summary": summary}

# === 子模式2：投票并行 ===

async def vote_parallel(question: str, n_votes: int = 3) -> str:
    """
    同一问题多次调用LLM（不同温度），通过聚合选最优
    适用于需要高准确性的判断任务
    """

    async def _vote(temperature: float) -> str:
        response = await aclient.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "请简洁回答问题。只输出答案，不要多余解释。"},
                {"role": "user", "content": question},
            ],
            temperature=temperature,
        )
        return response.choices[0].message.content.strip()

    # 不同温度并行调用，增加答案多样性
    temps = [0, 0.3, 0.7][:n_votes]
    answers = await asyncio.gather(*[_vote(t) for t in temps])

    print(f"🗳️ 投票结果: {answers}")

    # 聚合：让LLM综合判断最佳答案
    aggregate_resp = await aclient.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "以下是多个专家对同一问题的回答。请综合判断，选择或合成最准确的答案。"},
            {"role": "user", "content": f"问题: {question}\n\n" + "\n".join(f"专家{i+1}: {a}" for i, a in enumerate(answers))},
        ],
        temperature=0,
    )
    return aggregate_resp.choices[0].message.content

# 使用示例
async def main():
    # 分段并行
    analysis = await analyze_parallel("苹果公司2024年第四季度营收创历史新高，iPhone销量超预期增长15%。")
    print(f"📊 并行分析结果: {list(analysis.keys())}")

    # 投票并行
    best_answer = await vote_parallel("Python的GIL是什么？它对多线程有什么影响？")
    print(f"🏆 投票最优答案: {best_answer[:200]}...")

asyncio.run(main())  # asyncio.run：创建事件循环并执行顶层协程，是异步程序的标准入口点

"""
编排者-工作者模式（Orchestrator-Worker / Subagent模式）
编排者Agent动态分解任务 → 委派给专业Subagent → 汇总结果

参考：Anthropic "Building Effective Agents" (2025) — Orchestrator-Workers 模式
这也是 Claude Code、Cursor Agent、GitHub Copilot 等产品的核心架构
"""

from openai import OpenAI
import json
from typing import Callable

client = OpenAI()

class SubAgent:
    """子Agent：专注于单一领域的执行者"""

    def __init__(self, name: str, system_prompt: str, model: str = "gpt-4o"):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model

    def run(self, task: str) -> str:
        """执行子任务"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": task},
            ],
            temperature=0,
        )
        return response.choices[0].message.content

class Orchestrator:
    """编排者：动态分析任务、委派Subagent、汇总结果"""

    def __init__(self):
        self.subagents: dict[str, SubAgent] = {}

    def register(self, name: str, system_prompt: str, model: str = "gpt-4o"):
        """注册一个Subagent"""
        self.subagents[name] = SubAgent(name, system_prompt, model)

    def run(self, task: str) -> str:
        """编排者主流程"""
        agent_names = list(self.subagents.keys())
        agent_descriptions = "\n".join(f"- {name}" for name in agent_names)

        # Step 1：让LLM动态分解任务——决定需要哪些Subagent、做什么
        plan_resp = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": f"""你是一个任务编排专家。将用户任务分解为子任务，并分配给合适的Subagent。

可用的Subagent:
{agent_descriptions}

返回JSON格式：
{{"subtasks": [{{"agent": "agent名称", "task": "具体子任务描述"}}]}}"""},
                {"role": "user", "content": task},
            ],
            response_format={"type": "json_object"},
            temperature=0,
        )
        plan = json.loads(plan_resp.choices[0].message.content)
        subtasks = plan.get("subtasks", [])

        print(f"📋 编排者分解出 {len(subtasks)} 个子任务:")
        for st in subtasks:
            print(f"  → [{st['agent']}] {st['task']}")

        # Step 2：依次调用Subagent执行（也可改为并行）
        results = []
        for st in subtasks:
            agent_name = st["agent"]
            agent = self.subagents.get(agent_name)
            if not agent:
                results.append({"agent": agent_name, "result": f"❌ 未找到Subagent: {agent_name}"})
                continue

            print(f"\n🔧 [{agent_name}] 执行中...")
            result = agent.run(st["task"])
            results.append({"agent": agent_name, "task": st["task"], "result": result})
            print(f"  ✅ [{agent_name}] 完成")

        # Step 3：汇总所有Subagent的结果
        results_text = "\n\n".join(
            f"### {r['agent']}:\n{r['result']}" for r in results
        )

        synthesize_resp = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个高级分析师。请综合以下各专家的分析结果，生成一份完整、连贯的最终报告。"},
                {"role": "user", "content": f"原始任务：{task}\n\n各Subagent结果：\n{results_text}"},
            ],
            temperature=0,
        )
        return synthesize_resp.choices[0].message.content

# === 使用示例 ===

orchestrator = Orchestrator()

orchestrator.register("市场分析师", "你是一个市场分析专家。请从市场规模、竞争格局、增长趋势等角度分析。")
orchestrator.register("技术专家", "你是一个AI技术专家。请从技术可行性、架构方案、实现难度等角度分析。")
orchestrator.register("风险顾问", "你是一个风险管理专家。请识别潜在风险并提出缓解措施。")

report = orchestrator.run("评估在金融客服场景部署AI Agent的可行性")
print(f"\n📋 编排者最终报告:\n{report}")

"""
SimpleAgent: 一个完整的最简Agent框架
支持工具注册、ReAct推理、记忆管理
"""

from openai import OpenAI
import json
from typing import Callable, Any
from dataclasses import dataclass

@dataclass
class AgentConfig:
    """Agent配置"""
    model: str = "gpt-4o"
    max_steps: int = 10
    temperature: float = 0
    system_prompt: str = "你是一个有用的AI助手，可以使用工具来帮助用户完成任务。"

class SimpleAgent:
    """最简Agent框架"""

    def __init__(self, config: AgentConfig | None = None):
        self.config = config or AgentConfig()
        self.client = OpenAI()
        self.tools: dict[str, Callable] = {}
        self.tool_schemas: list[dict] = []
        self.conversation_history: list[dict] = []

    def tool(self, name: str, description: str, parameters: dict):
        """装饰器：注册工具"""
        def decorator(func: Callable):
            self.tools[name] = func
            self.tool_schemas.append({
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": parameters,
                },
            })
            return func
        return decorator

    def _execute_tool(self, name: str, arguments: dict) -> str:
        """执行工具调用"""
        if name not in self.tools:
            return json.dumps({"error": f"未知工具: {name}"})
        try:
            result = self.tools[name](**arguments)
            if isinstance(result, (dict, list)):
                return json.dumps(result, ensure_ascii=False)
            return str(result)
        except Exception as e:
            return json.dumps({"error": str(e)})

    def run(self, user_input: str) -> str:
        """运行Agent"""
        # 初始化消息
        if not self.conversation_history:
            self.conversation_history.append({
                "role": "system",
                "content": self.config.system_prompt,
            })

        self.conversation_history.append({
            "role": "user",
            "content": user_input,
        })

        for step in range(self.config.max_steps):
            # 调用LLM
            response = self.client.chat.completions.create(
                model=self.config.model,
                messages=self.conversation_history,
                tools=self.tool_schemas if self.tool_schemas else None,
                temperature=self.config.temperature,
            )

            message = response.choices[0].message

            # 无工具调用 → 返回结果
            if not message.tool_calls:
                self.conversation_history.append({
                    "role": "assistant",
                    "content": message.content,
                })
                return message.content

            # 有工具调用 → 执行工具
            self.conversation_history.append(message)

            for tool_call in message.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)

                print(f"  🔧 Step {step + 1}: {func_name}({json.dumps(func_args, ensure_ascii=False)})")

                result = self._execute_tool(func_name, func_args)

                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result,
                })

        return "达到最大执行步骤数"

    def reset(self):
        """重置对话历史"""
        self.conversation_history.clear()

# ===== 使用SimpleAgent =====

agent = SimpleAgent(AgentConfig(
    model="gpt-4o",
    max_steps=5,
    system_prompt="你是一个研究助手，可以搜索信息和进行计算。",
))

@agent.tool(
    name="web_search",
    description="搜索互联网获取最新信息",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索查询"},
        },
        "required": ["query"],
    },
)
def web_search(query: str) -> dict:
    """模拟网络搜索"""
    return {"results": [{"title": f"关于{query}的信息", "snippet": f"这是{query}的相关内容..."}]}

@agent.tool(
    name="calculator",
    description="进行数学计算",
    parameters={
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "数学表达式"},
        },
        "required": ["expression"],
    },
)
def calculator(expression: str) -> dict:
    """安全计算

    ⚠️ 安全提示: 生产环境请使用 simpleeval 库替代 eval():
        from simpleeval import simple_eval
        return {"result": simple_eval(expression)}
    """
    allowed_chars = set("0123456789+-*/.() ")
    # all(生成器): 检查每个字符是否都在白名单中
    if not all(c in allowed_chars for c in expression):
        return {"error": "不支持的字符"}
    try:
        return {"result": eval(expression)}  # noqa: S307 — 生产环境请替换为 simpleeval
    except Exception as e:
        return {"error": str(e)}

# 运行Agent
result = agent.run("帮我搜索一下GPT-5的最新消息，然后计算一下如果API价格降低30%，原价0.01美元/1k tokens现在是多少？")
print(f"\n📋 最终回答:\n{result}")