from typing import Callable, Dict, Any, Listclass Tool:    def __init__(self, name: str, description: str, function: Callable, parameters: Dict[str, Any]):        self.name = name        self.description = description        self.function = function        self.parameters = parametersclass ToolRegistry:    def __init__(self):        self.tools: Dict[str, Tool] = {}    def register(self, name: str, description: str, function: Callable, parameters: Dict[str, Any]):        tool = Tool(name, description, function, parameters)        self.tools[name] = tool    def get_tool(self, name: str) -> Tool:        return self.tools.get(name)    def list_tools(self) -> List[Tool]:        return list(self.tools.values())    def call_tool(self, name: str, **kwargs) -> Any:        tool = self.get_tool(name)        if tool is None:            raise ValueError(f"工具 {name} 未注册")        return tool.function(**kwargs)

import requestsimport jsondef get_weather(city: str) -> str:    url = f"http://api.weatherapi.com/v1/current.json?key=YOUR_API_KEY&q={city}"    response = requests.get(url)    if response.status_code == 200:        data = response.json()        return f"{city} 当前天气：{data['current']['temp_c']}度，{data['current']['condition']['text']}"    else:        return f"获取 {city} 天气失败"def search_web(query: str) -> str:    url = f"https://api.search.com?q={query}"    response = requests.get(url)    if response.status_code == 200:        results = response.json()["results"]        summaries = [result["summary"] for result in results[:3]]        return "".join(summaries)    else:        return "搜索失败"def calculate(expression: str) -> str:    try:        result = eval(expression)        return f"计算结果：{expression} = {result}"    except Exception as e:        return f"计算错误：{e}"tool_registry = ToolRegistry()tool_registry.register(    name="get_weather",    description="获取指定城市的当前天气信息",    function=get_weather,    parameters={        "city": {"type": "string", "description": "城市名称"}    })tool_registry.register(    name="search_web",    description="在互联网上搜索信息",    function=search_web,    parameters={        "query": {"type": "string", "description": "搜索查询词"}    })tool_registry.register(    name="calculate",    description="计算数学表达式",    function=calculate,    parameters={        "expression": {"type": "string", "description": "数学表达式"}    })

class FunctionCallParser:    def __init__(self, tool_registry: ToolRegistry):        self.tool_registry = tool_registry    def parse_function_call(self, model_output: str) -> Dict[str, Any]:        import re        pattern = r"\{\{\s*tool_call\s*:\s*\{(.*?)\}\}\}"        match = re.search(pattern, model_output, re.DOTALL)        if match:            try:                json_str = "{" + match.group(1) + "}"                return json.loads(json_str)            except json.JSONDecodeError:                return None        return None    def execute_function_call(self, function_call: Dict[str, Any]) -> Any:        tool_name = function_call.get("name")        arguments = function_call.get("arguments", {})        if tool_name is None:            raise ValueError("缺少工具名称")        return self.tool_registry.call_tool(tool_name, **arguments)    def process_model_output(self, model_output: str) -> str:        function_call = self.parse_function_call(model_output)        if function_call:            result = self.execute_function_call(function_call)            return f"工具调用结果：{result}"        return model_output

class ToolCallPrompt:    def __init__(self, tool_registry:ToolRegistry):        self.tool_registry = tool_registry    def build_prompt(self, user_query: str) -> str:        tools_description = []        for tool in self.tool_registry.list_tools():            params_desc = ", ".join([                f"{name} ({param['type']}): {param['description']}"                for name, param in tool.parameters.items()            ])            tools_description.append(f"- {tool.name}: {tool.description}（参数：{params_desc}）")        tools_text = "".join(tools_description)        prompt = f"""你是一个智能助手，可以使用以下工具：可用工具：{tools_text}工具调用格式：{{{{tool_call: {{"name": "工具名称", "arguments": {{"参数名": "参数值"}}}}}}}如果需要使用工具，请按照上述格式调用。如果不需要使用工具，可以直接回答问题。用户问题：{user_query}"""        return prompt

class Agent:    def __init__(self, name: str, role: str, model, tool_registry: ToolRegistry):        self.name = name        self.role = role        self.model = model        self.tool_registry = tool_registry        self.function_parser = FunctionCallParser(tool_registry)    def process(self, query: str, context: str = "") -> str:        prompt = f"""你是 {self.name}，你的角色是 {self.role}。上下文：{context}用户查询：{query}请根据你的角色回答问题或调用工具。"""        input_ids = self.model.tokenizer.encode(prompt, return_tensors="pt").cuda()        with torch.no_grad():            output = self.model.generate(input_ids, max_new_tokens=512)        response = self.model.tokenizer.decode(output[0], skip_special_tokens=True)        return self.function_parser.process_model_output(response)

class MultiAgentSystem:    def __init__(self, agents: List[Agent]):        self.agents = agents        self.agent_map = {agent.name: agent for agent in agents}    def route(self, query: str) -> Agent:        for agent in self.agents:            if agent.role in query or self._matches_agent_role(agent, query):                return agent        return self.agents[0]    def _matches_agent_role(self, agent: Agent, query: str) -> bool:        role_keywords = {            "财务分析师": ["财务", "报表", "利润", "收入", "成本"],            "法律顾问": ["法律", "合同", "法规", "合规", "条款"],            "技术专家": ["技术", "代码", "架构", "系统", "开发"],            "客服专员": ["客服", "问题", "帮助", "支持", "反馈"]        }        keywords = role_keywords.get(agent.role, [])        return any(keyword in query for keyword in keywords)    def solve(self, query: str) -> str:        agent = self.route(query)        print(f"路由到 Agent: {agent.name} ({agent.role})")        return agent.process(query)    def collaborative_solve(self, query: str) -> str:        results = []        for agent in self.agents:            print(f"Agent {agent.name} 正在处理...")            result = agent.process(query)            results.append({                "agent": agent.name,                "role": agent.role,                "result": result            })        summary_agent = self.agent_map.get("总结专家", self.agents[0])        context = "".join([            f"{r['agent']} ({r['role']}): {r['result']}"            for r in results        ])        summary_query = f"请综合以下各Agent的回答，给出最终总结：{context}"        return summary_agent.process(summary_query)

financial_agent = Agent(    name="财务分析师",    role="财务分析师",    model=deepseek_model,    tool_registry=tool_registry)legal_agent = Agent(    name="法律顾问",    role="法律顾问",    model=deepseek_model,    tool_registry=tool_registry)tech_agent = Agent(    name="技术专家",    role="技术专家",    model=deepseek_model,    tool_registry=tool_registry)multi_agent_system = MultiAgentSystem([financial_agent, legal_agent, tech_agent])query = "分析某公司的财务状况并评估其法律合规风险"result = multi_agent_system.collaborative_solve(query)print(result)

import requestsfrom datetime import datetimeclass OASystemConnector:    def __init__(self, api_url: str, token: str):        self.api_url = api_url        self.token = token        self.headers = {            "Authorization": f"Bearer {self.token}",            "Content-Type": "application/json"        }    def get_approval_status(self, request_id: str) -> Dict[str, Any]:        url = f"{self.api_url}/api/approval/{request_id}"        response = requests.get(url, headers=self.headers)        if response.status_code == 200:            return response.json()        return {"error": f"获取审批状态失败: {response.status_code}"}    def create_approval(self, data: Dict[str, Any]) -> Dict[str, Any]:        url = f"{self.api_url}/api/approval"        response = requests.post(url, headers=self.headers, json=data)        if response.status_code == 201:            return response.json()        return {"error": f"创建审批失败: {response.status_code}"}    def list_approvals(self, user_id: str, status: str = None) -> List[Dict[str, Any]]:        url = f"{self.api_url}/api/approvals?user_id={user_id}"        if status:            url += f"&status={status}"        response = requests.get(url, headers=self.headers)        if response.status_code == 200:            return response.json()        return []

class ERPSystemConnector:    def __init__(self, api_url: str, token: str):        self.api_url = api_url        self.token = token        self.headers = {            "Authorization": f"Bearer {self.token}",            "Content-Type": "application/json"        }    def get_inventory(self, product_id: str) -> Dict[str, Any]:        url = f"{self.api_url}/api/inventory/{product_id}"        response = requests.get(url, headers=self.headers)        if response.status_code == 200:            return response.json()        return {"error": f"获取库存失败: {response.status_code}"}    def create_order(self, order_data: Dict[str, Any]) -> Dict[str, Any]:        url = f"{self.api_url}/api/orders"        response = requests.post(url, headers=self.headers, json=order_data)        if response.status_code == 201:            return response.json()        return {"error": f"创建订单失败: {response.status_code}"}    def get_sales_data(self, start_date: str, end_date: str) -> Dict[str, Any]:        url = f"{self.api_url}/api/sales?start_date={start_date}&end_date={end_date}"        response = requests.get(url, headers=self.headers)        if response.status_code == 200:            return response.json()        return {"error": f"获取销售数据失败: {response.status_code}"}

def register_enterprise_tools(tool_registry: ToolRegistry, oa_connector: OASystemConnector, erp_connector: ERPSystemConnector):    tool_registry.register(        name="get_approval_status",        description="获取审批流程状态",        function=oa_connector.get_approval_status,        parameters={            "request_id": {"type": "string", "description": "审批单ID"}        }    )    tool_registry.register(        name="create_approval",        description="创建审批流程",        function=oa_connector.create_approval,        parameters={            "data": {"type": "object", "description": "审批数据"}        }    )    tool_registry.register(        name="get_inventory",        description="获取产品库存",        function=erp_connector.get_inventory,        parameters={            "product_id": {"type": "string", "description": "产品ID"}        }    )    tool_registry.register(        name="get_sales_data",        description="获取销售数据",        function=erp_connector.get_sales_data,        parameters={            "start_date": {"type": "string", "description": "开始日期(YYYY-MM-DD)"},            "end_date": {"type": "string", "description": "结束日期(YYYY-MM-DD)"}        }    )

oa_connector = OASystemConnector(    api_url="http://oa.company.com",    token="your_oa_token")erp_connector = ERPSystemConnector(    api_url="http://erp.company.com",    token="your_erp_token")enterprise_tool_registry = ToolRegistry()register_enterprise_tools(enterprise_tool_registry, oa_connector, erp_connector)enterprise_agent = Agent(    name="企业助手",    role="企业智能助手",    model=deepseek_model,    tool_registry=enterprise_tool_registry)query = "查询产品A123的库存情况，并查看我的审批单状态"result = enterprise_agent.process(query)print(result)