第 13 章 Agent 智能体源码开发与行业工具调用

13.1 Function Call 工具调用底层实现

13.1.1 工具注册与调用架构

from typing import Callable, Dict, Any, List

class 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 = parameters

class 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)

13.1.2 工具定义示例

import requests
import json

def 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": "数学表达式"}
    }
)

13.1.3 Function Call 解析与执行

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

13.1.4 工具调用提示词

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

13.2 多 Agent 协作流程源码改造

13.2.1 Agent 定义

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)

13.2.2 多 Agent 协作架构

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)

13.2.3 Agent 协作示例

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)

13.3 企业 OA/ERP 系统接口对接实战

13.3.1 OA 系统对接

import requests
from datetime import datetime

class 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 []

13.3.2 ERP 系统对接

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}"}

13.3.3 企业系统工具注册

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)"}
        }
    )

13.3.4 企业 Agent 实战

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)

本章小结：

本章详细介绍了 Agent 智能体的源码开发与行业工具调用，包括 Function Call 工具调用、多 Agent 协作流程和企业系统接口对接。这些技术为企业构建智能办公助手和自动化业务流程提供了完整的解决方案。
更多资讯:lxb20110121