企业官网建设流程全解析

2 智能体产品设计

隐私泄露；

歧视、危害国家安全。

设计时，根据诉求，发掘深层的意图，合规诉求。

展示，可以通过： 分步骤、需要对比的需要左中右分栏。

代码生成 智能体：分开发者、review视角。

语音交互设计

语音用户界面（VUI）设计原则：

1. 首轮简短，层级递进。 减少用户负担。

2. 听觉建议： 传达时，先传达结论，再传达细节。

3. 关键操作信息，复述并让客户确认。

4. 与视觉屏幕 + 触摸屏触觉结合，避免 嘈杂环境下不可操作。

输出：

1. 通过 首Token，降低用户等待焦虑。

2. 工具调用等，给出中间结果。避免用户以为没有响应。

打断：

1. 放弃TTS，继承上下文，立即开始新的意图理解。

播报： 断点续传机制。 继续播放时，可以选择段落开头，或者继续播放。

人机协同型智能体：

1. 中心化编排；先拆分，再 汇总。

2. 流水线结构；

3. 循环模式。

智能体触发； 智能体之间的交互规范；中间环节、报错的可见性。

AG-UI（Agent-User Interaction Protocol，智能体—用户交互协议）是一个开放、轻量、事件驱动的协议，用来标准化「后端 AI 智能体」与「前端用户界面」之间的实时双向通信。由 CopilotKit 团队于 2025 年 5 月开源。

解决的痛点很具体：过去做一个 Agent 应用，前端要自己用 WebSocket/SSE 手搓一套流式协议——怎么推 token、怎么显示"正在调用工具"、怎么同步 Agent 内部状态、怎么让用户中途打断或补充信息。

Agent 通过标准 HTTP/SSE（也支持 WebSocket、Webhook、二进制通道）向前端推送一个 JSON 事件序列，约 16 种标准事件类型，主要分四类：
消息事件（文本增量流式输出）
工具调用事件（工具名、参数、结果）
状态补丁事件（用 delta 增量同步 Agent 状态，而不是全量重传）
生命周期事件（run 开始/结束/报错）

3.1 智能体高代码开发

高代码允许 硬件级资源优化，允许自定义微服务粒度、通信协议、增加数据分片策略。能控制技术债务。

一个节点 同时完成很多任务的时候，成功率会迅速降低。 plan的长度越长，准确率也越低。

设计项目时：

1. 模块化分层：智能体系统应采用感知层、决策层、执行层的三层架构。 路由层、

2. 分层记忆机制：4层（工作、情景、语义、元信息）、记忆权限隔离、保留周期控制。
* 安全控制设计：实现输入过滤（防指令注入）、权限校验（API 白名单）、输出审查。

接口统一API规范、负载均衡。

对话历史内容过长的四种解决办法
转写、检索、压缩、隔离。

react模式的开发，对模型的要求高，研发成本高。 所以生产环境用工作流 速度最快、最稳定。

工作流的优化：1. 串行任务 并行化。 2. 模型微调 或者换模型。 3. 流式返回。

技术选型：

信息收集智能体

3.5 function call

模型不知道什么时候调用，怎么调用。

ds-3调用范例：

import json from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage from dotenv import load_dotenv import os import os import getpass # v2 版本：只需填写从千帆平台获取的 API Key # 获取路径：控制台 -> 安全认证 -> API Key -> 创建API Key（选择千帆ModelBuilder） qianfan_api_key = getpass.getpass("Please input QIANFAN API Key (格式: bce-v3/ALTAK-xxx/xxx):") os.environ["API_KEY"] = qianfan_api_key # 初始化大模型 llm = ChatOpenAI( model="deepseek-v3.2", temperature=0, api_key=os.getenv("API_KEY"), base_url="https://qianfan.baidubce.com/v2" ) print("大模型初始化成功!")

# 定义一个简单的加法函数 @tool def add(a: int, b: int) -> int: """将两个数字相加。 Args: a: 第一个数字 b: 第二个数字 Returns: 两数之和 """ return a + b # 将工具绑定到模型 llm_with_tools = llm.bind_tools([add]) print("工具定义完成!") print(f"工具名称: {add.name}") print(f"工具描述: {add.description}")

# 测试函数调用 query = "请帮我计算 123 加 456 等于多少?" messages = [HumanMessage(content=query)] response = llm_with_tools.invoke(messages) print("=== 模型响应 ===") print(f"是否调用工具: {bool(response.tool_calls)}") if response.tool_calls: tool_call = response.tool_calls[0] print(f"\n调用的工具: {tool_call['name']}") print(f"工具参数: {tool_call['args']}") # 执行实际的函数调用 result = add.invoke(tool_call['args']) print(f"\n执行结果: {result}")

=== 模型响应 ===

是否调用工具: True

调用的工具: add

工具参数: {'a': 123, 'b': 456}

执行结果: 579

def run_conversation(query: str, llm_with_tools, tools_dict: dict): """运行完整的对话流程""" print(f"用户提问: {query}\n") # 1. 发送用户消息 messages = [HumanMessage(content=query)] response = llm_with_tools.invoke(messages) # 2. 检查是否需要调用工具 if not response.tool_calls: print("模型直接回复:", response.content) return response.content # 3. 执行工具调用 messages.append(response) for tool_call in response.tool_calls: tool_name = tool_call['name'].strip() tool_args = tool_call['args'] print(f"调用工具: {tool_name}") print(f"参数: {tool_args}") # 执行工具 tool_result = tools_dict[tool_name].invoke(tool_args) print(f"工具返回: {tool_result}\n") # 将工具结果添加到消息历史 messages.append( ToolMessage( content=str(tool_result), tool_call_id=tool_call['id'] ) ) # 4. 获取最终回复 final_response = llm_with_tools.invoke(messages) print("最终回复:", final_response.content) return final_response.content

# 测试完整流程

tools_dict = {"add": add}

run_conversation(

"帮我算一下 789 + 321",

llm_with_tools,

tools_dict

)

@tool def multiply(a: float, b: float) -> float: """将两个数字相乘。 Args: a: 第一个数字 b: 第二个数字 """ return a * b @tool def divide(a: float, b: float) -> float: """将第一个数字除以第二个数字。 Args: a: 被除数 b: 除数 """ if b == 0: return "错误: 除数不能为0" return a / b @tool def power(base: float, exponent: float) -> float: """计算幂运算。 Args: base: 底数 exponent: 指数 """ return base ** exponent # 绑定所有工具 calculator_tools = [add, multiply, divide, power] llm_calculator = llm.bind_tools(calculator_tools) tools_dict = {t.name: t for t in calculator_tools} print(f"已注册 {len(calculator_tools)} 个工具:") for t in calculator_tools: print(f" - {t.name}: {t.description}")

@tool def get_current_weather(location: str, unit: str = "celsius") -> str: """获取指定城市的当前天气信息。 Args: location: 城市名称,例如 '北京', '上海', '深圳' unit: 温度单位, 'celsius'(摄氏度) 或 'fahrenheit'(华氏度) Returns: 天气信息的JSON字符串 """ # 模拟天气数据 weather_data = { "北京": {"temperature": 15, "condition": "晴朗", "humidity": 45}, "上海": {"temperature": 22, "condition": "多云", "humidity": 65}, "深圳": {"temperature": 28, "condition": "小雨", "humidity": 80}, "广州": {"temperature": 26, "condition": "阴天", "humidity": 70}, } if location in weather_data: data = weather_data[location].copy() if unit == "fahrenheit": data["temperature"] = data["temperature"] * 9/5 + 32 data["unit"] = "°F" else: data["unit"] = "°C" return json.dumps(data, ensure_ascii=False) else: return json.dumps({"error": f"未找到城市 {location} 的天气信息"}, ensure_ascii=False) @tool def get_weather_forecast(location: str, days: int = 3) -> str: """获取指定城市未来几天的天气预报。 Args: location: 城市名称 days: 预报天数(1-7天) Returns: 天气预报的JSON字符串 """ # 模拟预报数据 forecast = [] conditions = ["晴", "多云", "阴", "小雨"] for i in range(min(days, 7)): forecast.append({ "day": f"第{i+1}天", "temperature_high": 25 + i, "temperature_low": 15 + i, "condition": conditions[i % len(conditions)] }) return json.dumps({ "location": location, "forecast": forecast }, ensure_ascii=False) # 绑定天气工具 weather_tools = [get_current_weather, get_weather_forecast] llm_weather = llm.bind_tools(weather_tools) weather_tools_dict = {t.name: t for t in weather_tools} print("天气助手工具已准备好!") # 测试天气查询 weather_queries = [ # "北京现在天气怎么样?", # "上海和深圳哪个城市温度更高?", "告诉我广州未来3天的天气预报", ] for query in weather_queries: print("\n" + "="*60) run_conversation(query, llm_weather, weather_tools_dict) print("="*60)

# 测试并行调用 query = "请告诉我北京、上海和深圳这三个城市现在的天气情况" print(f"用户提问: {query}\n") messages = [HumanMessage(content=query)] response = llm_weather.invoke(messages) if response.tool_calls: print(f"模型决定并行调用 {len(response.tool_calls)} 个工具:\n") messages.append(response) # 执行所有工具调用 for i, tool_call in enumerate(response.tool_calls, 1): tool_name = tool_call['name'] tool_args = tool_call['args'] print(f"[{i}] 调用 {tool_name}") print(f" 参数: {tool_args}") result = weather_tools_dict[tool_name].invoke(tool_args) print(f" 结果: {result}\n") messages.append( ToolMessage( content=str(result), tool_call_id=tool_call['id'] ) ) # 获取最终回复 final_response = llm_weather.invoke(messages) print("\n最终回复:") print(final_response.content)

# 测试天气查询 weather_queries = [ # "北京现在天气怎么样?", # "上海和深圳哪个城市温度更高?", "告诉我广州未来3天的天气预报", ] for query in weather_queries: print("\n" + "="*60) run_conversation(query, llm_weather, weather_tools_dict) print("="*60)

============================================================

用户提问: 告诉我广州未来3天的天气预报

调用工具: get_weather_forecast

参数: {'location': '广州', 'days': 3}

工具返回: {"location": "广州", "forecast": [{"day": "第1天", "temperature_high": 25, "temperature_low": 15, "condition": "晴"}, {"day": "第2天", "temperature_high": 26, "temperature_low": 16, "condition": "多云"}, {"day": "第3天", "temperature_high": 27, "temperature_low": 17, "condition": "阴"}]}

最终回复: 根据天气预报，广州未来3天的天气情况如下：

**第1天：**

- 天气：晴

- 最高温度：25°C

- 最低温度：15°C

**第2天：**

- 天气：多云

- 最高温度：26°C

- 最低温度：16°C

**第3天：**

- 天气：阴

- 最高温度：27°C

- 最低温度：17°C

总体来说，广州未来3天天气逐渐转暖，温度在15-27°C之间，天气从晴天逐渐转为多云和阴天。建议您根据天气变化适当调整衣物。

============================================================

# 测试并行调用 query = "请告诉我北京、上海和深圳这三个城市现在的天气情况" print(f"用户提问: {query}\n") messages = [HumanMessage(content=query)] response = llm_weather.invoke(messages) if response.tool_calls: print(f"模型决定并行调用 {len(response.tool_calls)} 个工具:\n") messages.append(response) # 执行所有工具调用 for i, tool_call in enumerate(response.tool_calls, 1): tool_name = tool_call['name'] tool_args = tool_call['args'] print(f"[{i}] 调用 {tool_name}") print(f" 参数: {tool_args}") result = weather_tools_dict[tool_name].invoke(tool_args) print(f" 结果: {result}\n") messages.append( ToolMessage( content=str(result), tool_call_id=tool_call['id'] ) ) # 获取最终回复 final_response = llm_weather.invoke(messages) print("\n最终回复:") print(final_response.content)

3.7 大模型工具调用

mcp

mcp实质是带服务发现的function call。 模型与工具的适配，原来m*n的适配成本。现在是n个。

工具需要自主反思的机制。

SKILLS：

渐进式披露： 只看每一个的描述来决定是否选用。 选用的技能，才加载全部内容。 最后执行时，才去调用。

环境安装：pip install "fastmcp[all]"

mcp服务：1. tool:

2. 提示词模板： 一些环境信息

3. 资源的定义：

4. 辅助函数

服务端需要常驻运行：python mcp_server.py

客户端： 定义连接服务器的接口

包括：

1. 建立连接：

调用工具：

# 便捷方法 async def multiply(self, a, b): return await self.call("multiply", a=a, b=b) async def calculate(self, operation, a, b): return await self.call("calculate", operation=operation, a=a, b=b) async def analyze(self, text): return await self.call("text_analysis", text=text) async def time(self, format="readable"): return await self.call("get_current_time", format=format) # 使用示例 if __name__ == "__main__": async def test(): client = MCPClient() await client.connect() # 测试工具调用 result = await client.multiply(7, 8) print(f"7 * 8 = {result}") result = await client.calculate("add", 10, 20) print(f"10 + 20 = {result}") await client.disconnect() asyncio.run(test())

一般skill都是自动生成的。

skill生成包含以下几个东西：

1.

通用智能体如何实现：

swarm架构的去中心化，

1. 任务判定

2. 任务拆解，任务依赖关系。

3. while循环执行任务。调用具体的执行任务的函数。

while i < len(tasks)

任务的执行，管理，用：

from swarm import Swarm, Agent

swarm.run( agent=marketing_agent, messages=[{"role": "user", "content": "帮我策划Q4的市场活动，做一个完整的规划。"}], context_variables=context )

4.

代码检查任务：

1. 生成代码；

2. 测试代码： 一系列参数，exec执行。

3. 大模型修正代码。

4. 循环2/3，直到全部成功，或达到最大步数。

大模型调用函数：

1. 用大模型输出选择调用的函数，然后用正则解析，获取大模型想要调用的函数名。

2. 然后用字符串判定为真的方法，去调用函数。

大模型生成代码的执行：

1. 检查代码

2. 用python自带的，可以将字符串变为 函数名。

# 执行代码，提取函数 exec(code, {}, local_vars)

eval执行函数并获取值。

code = '2 + 2'
result = eval(code)

AI员工

需要构建一个Workspace。

一种尝试围绕Agent构建「团队研发仓库」的实践，一个让团队所有人都可以沉淀为「Agent可用资产」的载体。