企业级 Agent 案例¶

from typing import Literal
from pydantic import BaseModel
from openai import AsyncOpenAI  # OpenAI异步客户端，返回协程对象，适合高并发场景

# ---- 工具定义 ----
# 工具函数是Agent与外部系统交互的桥梁
# 每个工具封装一个独立的业务能力，Agent根据用户意图自主选择调用

# 订单信息数据模型，使用Pydantic做类型校验和序列化
class OrderInfo(BaseModel):
    order_id: str
    status: str        # 订单状态: pending(待处理) / shipped(已发货) / delivered(已送达) / refunded(已退款)
    product: str
    amount: float
    created_at: str

async def query_order(order_id: str) -> dict:
    """查询订单状态（接入企业订单系统）"""
    # 真实场景: 调用内部订单微服务，这里用模拟数据演示
    return {
        "order_id": order_id,
        "status": "shipped",
        "product": "无线耳机",
        "amount": 299.0,
        "tracking_number": "SF1234567890",
        "estimated_delivery": "2026-03-25"
    }

async def create_refund(order_id: str, reason: str) -> dict:
    """创建退款申请"""
    return {
        "refund_id": f"RF{order_id}",
        "status": "processing",
        "estimated_time": "3-5个工作日"
    }

async def search_faq(query: str) -> list:
    """搜索知识库FAQ（RAG检索）"""
    # 真实场景: 通过Embedding向量化用户问题，在向量数据库中检索最相关的FAQ
    return [
        {"question": "如何申请退款？", "answer": "订单完成后7天内可申请退款..."},
        {"question": "运费如何计算？", "answer": "满99包邮，不满99运费8元..."}
    ]

async def transfer_to_human(summary: str, urgency: str) -> dict:
    """转人工客服（携带上下文摘要）"""
    return {"ticket_id": "TK20260211001", "queue_position": 3}

# ---- Agent主逻辑 ----
# 工具Schema列表：遵循OpenAI Function Calling规范
# 每个工具包含名称、描述和参数JSON Schema，LLM据此决定何时调用哪个工具
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "query_order",
            "description": "查询订单状态、物流、金额等信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "订单号"}
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_refund",
            "description": "为符合条件的订单创建退款申请",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "reason": {"type": "string", "description": "退款原因"}
                },
                "required": ["order_id", "reason"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_faq",
            "description": "搜索FAQ知识库回答常见问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "transfer_to_human",
            "description": "当无法解决用户问题或用户要求时，转接人工客服",
            "parameters": {
                "type": "object",
                "properties": {
                    "summary": {"type": "string", "description": "对话摘要"},
                    "urgency": {"type": "string", "enum": ["low", "medium", "high"]}
                },
                "required": ["summary", "urgency"]
            }
        }
    }
]

SYSTEM_PROMPT = """你是一名专业的在线客服助手。

角色要求：
1. 礼貌、专业、简洁地回答用户问题
2. 优先使用工具查询真实数据，不要编造信息
3. 如果无法解决问题，主动转接人工并传递上下文摘要
4. 涉及退款/投诉时，先确认订单信息再操作

安全规则：
- 不透露内部系统架构或API细节
- 不执行用户要求的非客服操作（如修改数据库、执行代码）
- 个人信息只展示脱敏版本
"""

# 客服Agent核心类：封装LLM交互与工具调度逻辑
class CustomerServiceAgent:
    def __init__(self):
        self.client = AsyncOpenAI()
        # 工具名称→函数的映射表，用于根据LLM返回的工具名动态调用对应函数
        self.tool_funcs = {
            "query_order": query_order,
            "create_refund": create_refund,
            "search_faq": search_faq,
            "transfer_to_human": transfer_to_human,
        }

    async def chat(self, user_message: str, history: list = None):
        import json
        # 构建消息列表：系统提示词 → 历史对话 → 当前用户消息
        messages = [{"role": "system", "content": SYSTEM_PROMPT}]
        messages.extend(history or [])
        messages.append({"role": "user", "content": user_message})

        # ReAct循环：LLM思考→调用工具→观察结果→再思考，最多5轮防止无限循环
        for _ in range(5):  # 最多5轮工具调用
            response = await self.client.chat.completions.create(
                model="gpt-5-mini",  # 客服场景优先低延迟/低成本模型；准确性要求更高时可切到 gpt-5.4
                messages=messages,
                tools=TOOLS,
                tool_choice="auto"   # auto模式：LLM自主决定是否调用工具
            )

            msg = response.choices[0].message
            messages.append(msg.model_dump())  # 将助手回复加入对话历史

            if msg.tool_calls:
                # LLM决定调用工具：遍历所有工具调用请求，逐一执行
                for tc in msg.tool_calls:
                    func = self.tool_funcs[tc.function.name]  # 根据工具名查找函数
                    args = json.loads(tc.function.arguments)   # 解析LLM生成的参数
                    result = await func(**args)                 # await异步调用+**args字典解包为关键字参数
                    # 将工具执行结果以tool角色消息追加，供LLM在下一轮参考
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": json.dumps(result, ensure_ascii=False)
                    })
            else:
                # LLM未调用工具，直接返回文本回复给用户
                return msg.content

        return "抱歉，我暂时无法处理您的问题，正在为您转接人工客服。"

import json
from dataclasses import dataclass
from openai import AsyncOpenAI

@dataclass
class ReviewComment:
    file: str
    line: int
    severity: str       # critical / warning / suggestion
    category: str       # security / performance / style / logic
    message: str
    suggestion: str = ""

class CodeReviewAgent:
    """多维度代码审查Agent

    设计模式：并行多维度审查
    - 将审查任务拆分为安全、性能、逻辑等独立维度
    - 每个维度使用专用Prompt，并行执行提高效率
    - 最终按严重程度排序合并结果
    """

    # 审查维度配置：每个维度有独立的审查指令，指导LLM关注特定方面
    REVIEW_DIMENSIONS = {
        "security": """检查代码是否存在安全漏洞：
            - SQL注入、XSS、CSRF
            - 硬编码的密钥/密码
            - 不安全的反序列化
            - 路径遍历
            只报告diff中新增代码的问题。""",

        "performance": """检查性能问题：
            - N+1查询
            - 不必要的内存分配
            - 可优化的算法复杂度
            - 缺少索引的数据库查询
            只报告diff中新增代码的问题。""",

        "logic": """检查逻辑问题：
            - 边界条件处理
            - 错误处理是否完善
            - 并发/竞态条件
            - 资源泄露（未关闭的连接/文件）
            只报告diff中新增代码的问题。"""
    }

    async def review(self, diff: str) -> list[ReviewComment]:
        """并行执行多维度审查"""
        import asyncio

        # 为每个审查维度创建异步任务，实现并行审查
        tasks = [
            self._review_dimension(diff, dim, prompt)
            for dim, prompt in self.REVIEW_DIMENSIONS.items()
        ]

        # asyncio.gather并发执行所有维度的审查，显著降低总耗时
        results = await asyncio.gather(*tasks)

        # 合并所有维度的审查结果到统一列表
        all_comments = []
        for comments in results:
            all_comments.extend(comments)

        # 按严重程度排序：critical(必须修复) > warning(建议修复) > suggestion(可选优化)
        severity_order = {"critical": 0, "warning": 1, "suggestion": 2}
        # sort+lambda+dict.get：用字典映射将严重级别转为数值排序，未知级别默认为3（最低优先）
        all_comments.sort(key=lambda c: severity_order.get(c.severity, 3))

        return all_comments

    async def _review_dimension(self, diff, dimension, prompt) -> list:
        """单维度审查：使用专用Prompt让LLM聚焦特定审查方面"""
        client = AsyncOpenAI()

        # 代码审查默认使用 gpt-5.4，优先保证准确性和结构化输出稳定性
        # 配合 Structured Output 确保返回格式稳定可解析
        response = await client.chat.completions.create(
            model="gpt-5.4",
            messages=[
                {"role": "system", "content": f"你是一名资深代码审查专家，专注{dimension}审查。\n{prompt}"},
                {"role": "user", "content": f"请审查以下Git diff:\n\n{diff}"}
            ],
            # JSON Schema约束输出格式，确保LLM返回结构化审查结果
            response_format={
                "type": "json_schema",
                "json_schema": {
                    "name": "review_output",
                    "schema": {
                        "type": "object",
                        "properties": {
                            "comments": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "file": {"type": "string"},
                                        "line": {"type": "integer"},
                                        "severity": {"type": "string"},
                                        "message": {"type": "string"},
                                        "suggestion": {"type": "string"}
                                    }
                                }
                            }
                        }
                    }
                }
            }
        )

        result = json.loads(response.choices[0].message.content)
        # 将LLM返回的JSON审查结果映射为ReviewComment对象列表
        # category统一设为当前审查维度名；suggestion字段可选，用get设默认空字符串
        return [
            ReviewComment(
                file=c["file"],
                line=c["line"],
                severity=c["severity"],
                category=dimension,       # 自动标注所属审查维度（security/performance/logic）
                message=c["message"],
                suggestion=c.get("suggestion", "")  # 有些问题可能没有修改建议
            )
            for c in result.get("comments", [])  # 防御性取值，LLM未返回comments时不报错
        ]

class DataAnalysisAgent:
    """数据分析Agent: 自然语言 → SQL → 分析 → 可视化

    核心工作流：用户提问 → 理解表结构 → 生成SQL查询 → 执行分析代码 → 输出结论
    安全设计：SQL只读 + Python沙箱执行 + 查询超时限制
    """

    # Agent可用的三类工具：SQL查询、Python执行、表结构获取
    TOOLS = [
        {
            "type": "function",
            "function": {
                "name": "execute_sql",
                "strict": True,
                "description": "在只读数据库上执行SQL查询（仅支持SELECT）",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string", "description": "SELECT SQL语句"},
                        "database": {"type": "string", "description": "数据库名"}
                    },
                    "required": ["sql", "database"],
                    "additionalProperties": False
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "run_python",
                "strict": True,
                "description": "执行Python代码进行数据分析和可视化（可用pandas, matplotlib）",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "code": {"type": "string", "description": "Python代码"}
                    },
                    "required": ["code"],
                    "additionalProperties": False
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "get_schema",
                "strict": True,
                "description": "获取数据库表结构信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "database": {"type": "string"},
                        "table": {"type": "string", "description": "表名（传空字符串则返回所有表）"}
                    },
                    "required": ["database", "table"],
                    "additionalProperties": False
                }
            }
        }
    ]

    SYSTEM_PROMPT = """你是一名数据分析师Agent。

工作流程：
1. 理解用户的分析需求
2. 先用get_schema了解数据结构
3. 用execute_sql查询所需数据（只允许SELECT）
4. 用run_python对查询结果进行分析和可视化
5. 给出清晰的分析结论和建议

规则：
- SQL只能执行SELECT语句
- 大表查询时主动加LIMIT（默认1000行）
- 图表要有中文标题和标签
- 分析结论要包含具体数字"""

    async def execute_sql(self, sql: str, database: str) -> str:
        """安全的SQL执行（只允许SELECT）"""
        # 安全防线1：白名单校验，只允许SELECT语句，防止数据篡改
        sql_stripped = sql.strip().upper()
        if not sql_stripped.startswith("SELECT"):
            return "错误：只允许执行SELECT查询"

        # 安全防线2：使用只读副本连接，即使SQL注入也无法写入数据
        import aiomysql
        # async with：异步上下文管理器，自动管理数据库连接的建立和释放，防止连接泄漏
        async with aiomysql.connect(
            host="readonly-replica.db.internal",  # 只读副本，非主库
            db=database,
            user="readonly_user",                  # 只读权限账号
            read_timeout=30                        # 30秒超时防止慢查询
        ) as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                await cur.execute(sql)
                rows = await cur.fetchall()
                # 将查询结果序列化为JSON字符串，供LLM解读分析
                return json.dumps(rows, default=str, ensure_ascii=False)

    async def run_python(self, code: str) -> str:
        """沙箱化Python执行：在隔离环境中运行LLM生成的代码

        ⚠️ 安全警告：当前实现仅为基础隔离，生产环境务必使用以下方案之一：
        - Docker容器沙箱（推荐）：限制网络/文件系统/内存/CPU
        - E2B (e2b.dev)：云端安全沙箱服务
        - gVisor/nsjail：内核级沙箱隔离
        仅用subprocess无法防止文件系统读写、网络访问等风险。
        """
        # 生产环境: 必须使用Docker沙箱或E2B，限制文件系统/网络访问
        # 此处subprocess仅为演示，不构成安全沙箱！
        import subprocess
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True,   # 捕获stdout和stderr，不直接输出到终端
            text=True,             # 以文本模式返回结果（而非bytes）
            timeout=30,            # 30秒超时，防止死循环或恶意代码长时间占用资源
            cwd="/tmp",            # 限制工作目录到临时目录
            env={"PATH": "/usr/bin"}  # 最小化环境变量，限制可执行程序访问范围
        )
        # 优先返回标准输出；若为空则返回错误信息，便于LLM诊断并修正代码
        return result.stdout or result.stderr

class ResearchAgent:
    """深度研究Agent: 搜索 → 阅读 → 综合 → 报告

    设计思路：模拟人类研究员的工作流程
    1. 广泛搜索 → 2. 深度阅读 → 3. 交叉验证 → 4. 综合报告
    使用多种信息源（网页、学术论文）确保研究质量
    """

    # Agent工具集：覆盖信息获取的完整链路
    TOOLS = [
        # 1. 网络搜索：获取最新公开信息
        {
            "type": "function",
            "function": {
                "name": "web_search",
                "description": "搜索最新信息（论文、新闻、博客）",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"},
                        "num_results": {"type": "integer", "default": 10}
                    },
                    "required": ["query"]
                }
            }
        },
        # 2. 页面阅读：获取搜索结果中网页的完整内容，用于深度分析和事实提取
        {
            "type": "function",
            "function": {
                "name": "read_webpage",
                "description": "读取网页全文内容",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "url": {"type": "string"}
                    },
                    "required": ["url"]
                }
            }
        },
        # 3. 论文搜索：通过学术搜索引擎检索同行评审论文，确保关键论点有权威来源
        {
            "type": "function",
            "function": {
                "name": "search_papers",
                "description": "在Semantic Scholar搜索学术论文",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"},
                        "year_from": {"type": "integer"}
                    },
                    "required": ["query"]
                }
            }
        },
        # 4. 生成报告：将多来源信息整合为带引用的结构化研究报告
        {
            "type": "function",
            "function": {
                "name": "write_report",
                "description": "将收集的信息整合成结构化研究报告",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "title": {"type": "string"},
                        # 报告由多个章节组成，每章包含标题、正文和引用来源列表
                        "sections": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "heading": {"type": "string"},   # 章节标题
                                    "content": {"type": "string"},   # 章节正文内容
                                    "sources": {"type": "array", "items": {"type": "string"}}  # 引用来源URL列表
                                }
                            }
                        }
                    },
                    "required": ["title", "sections"]
                }
            }
        }
    ]

    SYSTEM_PROMPT = """你是一名专业研究助手。

工作流程：
1. 先搜索广泛信息，了解领域全貌
2. 深入阅读最相关的3-5篇文章/论文
3. 综合多个来源，交叉验证关键事实
4. 生成结构化报告（含引用来源）

质量标准:
- 每个关键论点至少有2个独立来源支撑
- 明确区分"事实"和"观点"
- 标注信息的时效性
- 如有争议观点，展示不同立场"""

import json
import hashlib
from datetime import datetime

class AgentGovernance:
    """企业Agent治理：权限、审计、合规

    企业级Agent必须具备的三大治理能力：
    1. 权限分级 - 不同角色可调用不同工具，遵循最小权限原则
    2. 操作审计 - 所有Agent操作留痕，支持事后追溯
    3. 合规检查 - 自动检测Agent输出是否符合合规要求
    """

    # 1. 权限分级：基于角色的访问控制(RBAC)
    # 每个级别定义可用工具、SQL操作权限和最大迭代次数
    PERMISSION_LEVELS = {
        "read_only": {  # 只读用户：仅能查询，不能执行写操作
            "allowed_tools": ["search_faq", "query_order", "get_schema", "execute_sql"],
            "sql_operations": ["SELECT"],
            "max_iterations": 5,
        },
        "standard": {  # 普通客服：可查询+创建退款，不能直接操作数据库
            "allowed_tools": ["search_faq", "query_order", "create_refund", "execute_sql"],
            "sql_operations": ["SELECT"],
            "max_iterations": 10,
        },
        "admin": {  # 管理员：拥有所有工具权限，可执行数据库写操作
            "allowed_tools": ["*"],  # 所有工具
            "sql_operations": ["SELECT", "INSERT", "UPDATE"],
            "max_iterations": 20,
        }
    }

    # 2. 操作审计：记录Agent每一次操作，确保可追溯
    async def audit_log(self, event):
        """所有Agent操作写入审计日志"""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),  # UTC时间戳
            "agent_id": event["agent_id"],                # Agent实例ID
            "user_id": event["user_id"],                  # 操作用户ID
            "action": event["action"],                    # 操作类型
            "tool_name": event.get("tool_name"),          # 调用的工具名
            # 对工具参数做哈希处理，记录操作指纹但不存储敏感原始数据
            "tool_args_hash": hashlib.sha256(
                json.dumps(event.get("tool_args", {})).encode()
            ).hexdigest(),
            "result_status": event["status"],             # 执行结果状态
            "cost_estimate": event.get("cost"),           # 本次调用成本估算
        }
        # 写入不可变审计存储（如append-only日志），防止事后篡改
        await self.audit_store.append(log_entry)

    # 3. 合规检查：对Agent输出进行自动化合规校验
    def compliance_check(self, user_query, agent_response):
        """检查Agent响应是否合规"""
        checks = {
            "pii_check": not self._contains_pii(agent_response),                    # 检查是否泄露个人隐私信息
            "no_financial_advice": not self._contains_financial_advice(agent_response),  # 检查是否含投资建议
            "source_cited": self._has_sources(agent_response),                       # 检查是否引用了来源
        }
        # 所有检查项通过才算合规
        return all(checks.values()), checks  # all()对字典所有值求逻辑与；返回元组(是否全部通过, 各项详情)

"""旅行助手Agent —— 完整实现"""

from dataclasses import dataclass
from openai import OpenAI

client = OpenAI()

# ────── 工具定义 ──────

# 景点数据库（模拟数据，生产环境替换为真实API或数据库）
# 数据结构：城市名 → 景点列表，每个景点含类型、费用、游览时长、评分
ATTRACTIONS_DB = {
    "成都": [
        {"name": "大熊猫基地", "type": "景点", "cost": 55, "hours": 3, "rating": 4.8},
        {"name": "宽窄巷子", "type": "景点", "cost": 0, "hours": 2, "rating": 4.5},
        {"name": "锦里", "type": "美食街", "cost": 0, "hours": 2, "rating": 4.3},
        {"name": "武侯祠", "type": "历史", "cost": 50, "hours": 2, "rating": 4.6},
        {"name": "都江堰", "type": "景点", "cost": 80, "hours": 4, "rating": 4.7},
        {"name": "青城山", "type": "自然", "cost": 90, "hours": 5, "rating": 4.6},
        {"name": "春熙路", "type": "购物", "cost": 0, "hours": 2, "rating": 4.2},
        {"name": "人民公园", "type": "休闲", "cost": 0, "hours": 1.5, "rating": 4.4},
    ],
    "北京": [
        {"name": "故宫", "type": "历史", "cost": 60, "hours": 4, "rating": 4.9},
        {"name": "长城(八达岭)", "type": "景点", "cost": 40, "hours": 6, "rating": 4.8},
    ],
}

def search_attractions(city: str, interest: str = "") -> str:
    """查询城市景点信息"""
    # 从景点数据库获取城市景点列表，未收录的城市返回空列表（优雅降级，不抛异常）
    attractions = ATTRACTIONS_DB.get(city, [])
    if interest:
        # 按兴趣类型过滤景点，支持模糊匹配（如"历史"匹配"历史"类型）
        # 列表推导+in子串匹配：过滤type字段包含interest的景点（如"历史"匹配"历史文化"类型）
        attractions = [a for a in attractions if interest in a["type"]]
    if not attractions:
        return f"未找到{city}的{'相关' if interest else ''}景点信息"
    # 格式化输出：每行一个景点，包含类型、费用、游览时长、评分
    # 结构化文本便于LLM提取信息并组织进行程规划
    lines = [f"📍 {city}景点推荐:"]
    for a in attractions:
        lines.append(f"  • {a['name']} ({a['type']}) - ¥{a['cost']} | {a['hours']}h | ⭐{a['rating']}")
    return "\n".join(lines)

def check_weather(city: str, month: int) -> str:
    """查询城市月份天气概况（模拟数据）"""
    # 使用(城市, 月份)元组作为字典键，实现多维度精确查找
    # 生产环境应替换为天气API实时查询（如和风天气、OpenWeatherMap）
    weather_data = {
        ("成都", 5): "5月成都：平均20-28°C，多云为主，偶有阵雨，建议携带薄外套和雨具",
        ("成都", 10): "10月成都：平均15-22°C，秋高气爽，适合出行",
        ("北京", 5): "5月北京：平均16-28°C，晴天为主，早晚温差大",
    }
    # dict.get兜底：未命中的城市/月份组合返回通用提示，避免返回空值
    return weather_data.get((city, month), f"{month}月{city}：预计温度适宜，建议出行前查看实时预报")

def calculate_budget(items: list[dict]) -> str:
    """计算预算明细：汇总各项费用并生成格式化账单"""
    total = 0
    lines = ["💰 预算明细:"]
    for item in items:
        cost = item.get("cost", 0)  # 使用get设默认值0，防止字段缺失
        total += cost
        lines.append(f"  • {item['name']}: ¥{cost}")
    lines.append(f"  ── 合计: ¥{total}")
    return "\n".join(lines)

# ────── Agent工具Schema ──────

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_attractions",
            "description": "查询指定城市的景点信息，可按类型过滤",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名"},
                    "interest": {"type": "string", "description": "兴趣类型：景点/历史/美食街/自然/购物/休闲"},
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "check_weather",
            "description": "查询城市指定月份的天气概况",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "month": {"type": "integer", "description": "月份(1-12)"},
                },
                "required": ["city", "month"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_budget",
            "description": "计算旅行预算，传入费用项列表",
            "parameters": {
                "type": "object",
                "properties": {
                    "items": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "name": {"type": "string"},
                                "cost": {"type": "number"},
                            },
                        },
                    },
                },
                "required": ["items"],
            },
        },
    },
]

# 工具名称→函数的映射表，Agent主循环通过此映射动态分派工具调用
TOOL_MAP = {
    "search_attractions": search_attractions,
    "check_weather": check_weather,
    "calculate_budget": calculate_budget,
}

# 系统提示词：定义Agent的角色、工作流程和行为约束
SYSTEM_PROMPT = """你是一位专业的旅行规划助手。你的职责：
1. 根据用户的目的地、天数、预算，规划合理的行程
2. 先查天气，再推荐景点，最后计算预算
3. 输出结构化的每日行程安排
4. 确保总费用不超过用户预算
5. 给出实用的旅行小贴士"""

def travel_agent(user_request: str) -> str:
    """旅行助手Agent主循环：接收用户自然语言请求，返回完整旅行规划"""
    import json

    # 初始化消息列表：系统提示词设定Agent角色 + 用户请求
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_request},
    ]

    # Agent主循环：LLM思考→调用工具→观察结果→继续思考
    # 旅行规划通常需要3-5轮（查天气→查景点→算预算），设置8轮上限
    for _ in range(8):  # 最大8轮工具调用
        response = client.chat.completions.create(
            model="gpt-5-mini",   # 旅行规划默认用低延迟/低成本模型，复杂约束场景可升级到 gpt-5.4
            messages=messages,
            tools=TOOLS,
        )
        msg = response.choices[0].message

        if not msg.tool_calls:
            return msg.content  # 无工具调用表示Agent已完成规划，返回最终行程

        # LLM请求调用工具：依次执行每个工具调用并将结果反馈给LLM
        messages.append(msg.model_dump())
        for tc in msg.tool_calls:
            func = TOOL_MAP.get(tc.function.name)  # 从映射表查找工具函数
            if func:
                args = json.loads(tc.function.arguments)  # 解析LLM生成的参数
                result = func(**args)                      # 执行工具
            else:
                result = f"未知工具: {tc.function.name}"   # 防御性处理
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result,  # 工具结果作为tool角色消息追加
            })

    return "规划轮次超限，请简化需求后重试"  # 超过最大轮次的兜底响应

# 使用示例
if __name__ == "__main__":
    plan = travel_agent("帮我规划5月去成都3天的旅行，预算5000元，喜欢美食和历史文化")
    print(plan)