民法典app官方版
21.56MB · 2025-10-09
从22年底ChatGPT的爆火，到今年年初国产大语言模型DeepSeekv3的频繁刷屏，我们看到了大语言模型（LLM）那令人惊叹的能力：写诗、编程、答疑仿佛无所不能，无所不知。其高度自然的人机交互界面，容易让我们感觉像是在与一位“全知全能”的对话者交流。然而，揭开这层表象，LLM的核心本质是一个基于海量数据训练的“下一个词元概率预测模型”。
他的核心任务非常简单：就是根据上下文（即你的问题）从数十亿的参数中，计算出下一个最可能出现的词元是什么。如此不断反复最终生成一个长文本。
当然如果仅仅是一个简单的下一个词元的概率预测模型，LLM也不会像现在这样火热。关键在于，凭借其庞大的参数规模与训练数据量，LLM显现了“涌现能力”（Emergent Abilities）也就是我们的俗语书读百遍，其意自现——这意味着当模型规模超过某个临界点时，它会展现出诸如复杂推理、逻辑分析、任务规划等并未在训练中被明确指定的新能力。
因此，LLM的能力呈现出一种二元性：
正是为了克服LLM的固有局限，同时充分利用其强大的推理和规划能力，Prompt工程、检索增强生成（RAG）、函数调用（Function Calling）和模型上下文协议（MCP）等技术应运而生。它们共同协作，构成了当今火热无比的AI Agent（智能代理）。
如下图所示：
我们可以将AI Agent想象为一个高效的超级助理：
接下来，我们将初步了解Prompt、RAG、Function Calling和MCP这四大关键技术，看看他们是如何工作的，在**下一篇文章中，**我将以一个实际的形同为例，带你看看它们如何协同工作，将一个简单的LLM武装成一个“数据超级助理”。
prompt（提示词）不是一种具体的技术，是我们和LLM沟通的控制语言。他的构成就是指令式的语言，这是整个Agent中的一个核心，我们所有功能的实现都是基于一个我们精心设计好的提示词。
提示词的核心作用就玩一场cosplay。我们要精确的告诉模型“你是谁（角色）”、“你要做什么（任务）”、“依照什么样的规则（约束）”、“参考什么信息（上下文，非必须）”，以及示例（非必须）。
LLM模型接受的训练数据是多种多样的，所以在回答时他的答案边界就会很模糊，Prompt的作用就是给模型画个圈，让他的答案限制在这个圈内。
接下来让我们做一个简单的演示，我们让模型来写一个关于狗的故事。
>(模型一) 写一个300字左右的关于狗的故事：
>(模型二) 你是一位小说家，你现在要用欧亨利的风格写一个300字左右的关于狗的故事。
我们用腾讯元宝中的DeepSeek，开启深度思考，会有如下的结果：
可以看出，两篇故事还是有很大的区别的。后一篇明显可以看出欧亨利的意料之外情理之中的结尾。
那么我们应该如何写好一个提示词呢？
在实践中，我们可以用上面的基本原则加我们下面的基本公式来生成你的提示词：
小Tips：
RAG（Retrieval-Augmented Generation，检索增强生成）：是一种将信息检索和LLM结合的技术框架，让LLM在回答问题前，先从外部知识库（内部数据库、PDF）获取相关信息然后将查询到的信息注入到Prompt中，利用LLM的总结能力来回答问题。
如图所示，RAG的调用流程共包含5个步骤
环境准备
pip install openai scikit-learn python-dotenv
在项目文件夹中创建一个.env的文件，填入你的LLM API
这里使用硅基流动的API，有免费的模型可以使用：Qwen/Qwen3-8B 这是我的邀请链接大家需要的话可以点一点，硅基流动现在学生注册免费送50元额度： cloud.siliconflow.cn/i/elkOMWfK
API_KEY的生成方式为：登陆硅基流动，在左边点击API密钥，点击新建密钥，粘贴到SILICONFLOW_API_KEY中就可以。
# SiliconFlow 的 API 地址
export SILICONFLOW_BASE_URL="https://api.siliconflow.cn/v1"
# 您在 SiliconFlow 平台获取的 API Key
export SILICONFLOW_API_KEY="sk-..."
# rag_complete.py
import os
import asyncio
from openai import AsyncOpenAI
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from dotenv import load_dotenv
# --- 1. 设置环境 ---
load_dotenv() # 加载 .env 文件中的环境变量
# 从环境变量中获取配置
BASE_URL = os.getenv("SILICONFLOW_BASE_URL")
API_KEY = os.getenv("SILICONFLOW_API_KEY")
if not API_KEY or not BASE_URL:
print("错误: 请确保 SILICONFLOW_BASE_URL 和 SILICONFLOW_API_KEY 环境变量已设置。")
exit()
# 初始化一个指向指定端点的异步 OpenAI 客户端
llm_client = AsyncOpenAI(
base_url=BASE_URL,
api_key=API_KEY
)
# --- 2. 构建简易知识库 ---
knowledge_base = [
"Starlight-15 是 Starlight Inc. 公司于 2024 年发布的旗舰智能手机，配备了超感光摄像头和 A18 Pro 芯片。",
"Galaxy Fold 6 是三星公司推出的最新折叠屏手机，主打大屏体验和多任务处理能力。",
"Starlight Inc. 公司的售后服务政策是自购买之日起一年内免费维修非人为损坏。",
"A18 Pro 芯片采用了 3 纳米工艺，集成了 200 亿个晶体管，性能比上一代提升了 20%。",
"Starlight-15 的电池容量为 5000mAh，支持 80W 快充，30分钟可充满 100%。"
]
vectorizer = TfidfVectorizer()
knowledge_base_vectors = vectorizer.fit_transform(knowledge_base)
# --- 3. 实现检索器 ---
def retrieve_context(query, top_k=2):
"""根据用户问题，从知识库中检索最相关的 K 个文本片段。"""
query_vector = vectorizer.transform([query])
similarities = cosine_similarity(query_vector, knowledge_base_vectors).flatten()
top_k_indices = similarities.argsort()[-top_k:]
return [knowledge_base[i] for i in reversed(top_k_indices)]
# --- 4. RAG 核心流程 ---
async def simple_rag_query(query):
"""执行完整的 RAG 流程"""
print(f"用户问题: {query}")
retrieved_context = retrieve_context(query)
print(f"检索到的上下文:n- {'n- '.join(retrieved_context)}")
augmented_prompt = (
"你是一个问答机器人。请根据以下提供的上下文信息来回答用户的问题。n"
"如果上下文信息不足以回答问题，请说'根据我所掌握的信息，无法回答该问题'。nn"
"--- 上下文 ---n"
f"{'n'.join(retrieved_context)}n"
"--- 问题 ---n"
f"{query}n"
"--- 回答 ---n"
)
print("n增强后的 Prompt (发送给 LLM 的内容):")
print("-----------------------------------------")
print(augmented_prompt)
print("-----------------------------------------")
try:
response = await llm_client.chat.completions.create(
model="Qwen/Qwen3-8B",
messages=[{"role": "user", "content": augmented_prompt}]
)
final_answer = response.choices[0].message.content
print(f"nLLM 生成的最终答案:n{final_answer}")
except Exception as e:
print(f"n调用 LLM 时发生错误: {e}")
async def main():
await simple_rag_query("Starlight-15 用的什么芯片？充电快吗？")
print("n" + "="*50 + "n")
await simple_rag_query("三星的折叠屏手机怎么样？")
if __name__ == "__main__":
asyncio.run(main())
Function Calling (函数调用) 让LLM能够调用外部的API或代码机制。这是模型与现实世界互动的“双手”。LLM 负责“决定调用哪个函数”，而外部程序负责“真正执行”。
# function_calling_complete.py
import os
import asyncio
import json
from openai import AsyncOpenAI
from dotenv import load_dotenv
# --- 1. 设置环境 ---
load_dotenv()
BASE_URL = os.getenv("SILICONFLOW_BASE_URL")
API_KEY = os.getenv("SILICONFLOW_API_KEY")
if not API_KEY or not BASE_URL:
print("错误: 请确保 SILICONFLOW_BASE_URL 和 SILICONFLOW_API_KEY 环境变量已设置。")
exit()
llm_client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
# --- 2. 定义我们的 Python 函数 (工具) ---
def get_current_weather(location: str, unit: str = "celsius"):
"""获取指定地点的当前天气信息。"""
print(f"--- 正在执行函数: get_current_weather(location='{location}', unit='{unit}') ---")
if "nanjing" in location.lower() or "南京" in location.lower():
return json.dumps({"location": "南京", "temperature": "15", "unit": unit, "forecast": "晴天"})
else:
return json.dumps({"location": location, "temperature": "未知"})
def send_email(to: str, subject: str, body: str):
"""向指定收件人发送邮件。"""
print(f"--- 正在执行函数: send_email(to='{to}', subject='{subject}', body='...') ---")
return json.dumps({"status": "成功", "message": f"邮件已成功发送至 {to}。"})
available_functions = {
"get_current_weather": get_current_weather,
"send_email": send_email,
}
# --- 3. Function Calling 核心流程 ---
async def run_conversation(user_query):
"""执行完整的函数调用流程"""
print(f"用户问题: {user_query}")
messages = [{"role": "user", "content": user_query}]
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "获取一个指定地点的当前天气情况",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "城市名称, 例如：南京"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位"},
}, "required": ["location"],
},
},
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "向指定收件人发送一封邮件",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人的邮箱地址"},
"subject": {"type": "string", "description": "邮件的主题"},
"body": {"type": "string", "description": "邮件的正文内容"},
}, "required": ["to", "subject", "body"],
},
}
}
]
try:
response = await llm_client.chat.completions.create(
model="Qwen/Qwen3-8B",
messages=messages,
tools=tools,
tool_choice="auto",
)
response_message = response.choices[0].message
if response_message.tool_calls:
print("LLM 决定调用一个工具。")
messages.append(response_message)
for tool_call in response_message.tool_calls:
function_name = tool_call.function.name
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
print(f" - 函数名: {function_name}, 参数: {function_args}")
function_response = function_to_call(**function_args)
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
})
second_response = await llm_client.chat.completions.create(
model="Qwen/Qwen3-8B", messages=messages
)
final_answer = second_response.choices[0].message.content
else:
print("LLM 决定不调用工具，直接回答。")
final_answer = response_message.content
print(f"nLLM 生成的最终答案:n{final_answer}")
except Exception as e:
print(f"n与 LLM 交互时发生错误: {e}")
async def main():
await run_conversation("南京现在天气怎么样？")
print("n" + "="*50 + "n")
await run_conversation("你好吗？")
if __name__ == "__main__":
asyncio.run(main())
如果RAG和Function Calling是“工具”，那么Agent就是那个总指挥。Agent不是一个单一的技术，而是一套架构的范式，其核心就是利用LLM强大的推理能力来自主规划、分解任务并决定调用哪个工具。
ReAct范式模拟了人类解决问题的方式：
也即形成一个如下图的循环。
# agent_complete.py
import os
import asyncio
import json
import re
from openai import AsyncOpenAI
from dotenv import load_dotenv
# --- 1. 设置环境 ---
load_dotenv()
BASE_URL = os.getenv("SILICONFLOW_BASE_URL")
API_KEY = os.getenv("SILICONFLOW_API_KEY")
if not API_KEY or not BASE_URL:
print("错误: 请确保 SILICONFLOW_BASE_URL 和 SILICONFLOW_API_KEY 环境变量已设置。")
exit()
llm_client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
# --- 2. 定义 Agent 可以使用的工具 ---
def get_current_weather(location: str):
"""获取指定地点的当前天气信息。"""
print(f"--- 正在执行工具: get_current_weather(location='{location}') ---")
if "nanjing" in location.lower() or "南京" in location.lower():
return "南京当前温度为15摄氏度，天气晴朗。"
else:
return f"抱歉，我没有 {location} 的天气信息。"
def search_web(query: str):
"""在网上搜索信息。"""
print(f"--- 正在执行工具: search_web(query='{query}') ---")
if "室内活动" in query:
return "南京热门的室内活动包括参观博物院、去新街口购物。"
elif "户外活动" in query:
return "南京晴天时热门的户外活动包括游览中山陵、逛玄武湖公园。"
else:
return f"关于 '{query}' 的搜索结果在此模拟中不可用。"
available_tools = {
"get_current_weather": get_current_weather,
"search_web": search_web,
}
# --- 3. 定义 Agent 的核心 Prompt ---
AGENT_SYSTEM_PROMPT = """
你是一个作为自主智能体运行的得力助手。
你的目标是通过将用户的请求分解成一系列步骤来解决它。
在每一步，你都必须遵循以下格式：
Thought: [你的推理和下一步行动的计划。分析上一步的观察结果并决定接下来做什么。]
Action: [一个 JSON 对象，代表你想要使用的工具。它必须是可用工具之一，或者是用于最终回答的工具。]
可用工具:
- `get_current_weather`:
- 描述: 获取指定地点的当前天气。
- 参数: `{ "location": "string" }`
- `search_web`:
- 描述: 在网上搜索信息。
- 参数: `{ "query": "string" }`
当你拥有足够信息来回答用户的问题时，你必须使用 `final_answer` 工具。
- `final_answer`:
- 描述: 向用户提供最终的答案。
- 参数: `{ "answer": "string" }`
单步示例:
Thought: 我需要找出南京的天气。我将使用 get_current_weather 工具。
Action: { "tool_name": "get_current_weather", "arguments": { "location": "南京" } }
"""
# --- 4. Agent 的主循环 ---
async def run_agent(user_query, max_steps=5):
"""执行 Agent 的思考-行动循环"""
print(f"Agent 任务开始: {user_query}n")
messages = [
{"role": "system", "content": AGENT_SYSTEM_PROMPT},
{"role": "user", "content": f"请解决以下请求: {user_query}"}
]
for step in range(max_steps):
print(f"--- 步骤 {step + 1} ---")
try:
response = await llm_client.chat.completions.create(
model="Qwen/Qwen3-8B",
messages=messages,
temperature=0,
)
response_text = response.choices[0].message.content
print(response_text)
thought_match = re.search(r"Thought: (.*?)nAction: ", response_text, re.DOTALL)
action_match = re.search(r"Action: ({.*}|{.*?s*.*?})", response_text, re.DOTALL)
if not thought_match or not action_match:
print("解析失败: 无法找到 Thought 或 Action。")
return
action_json_str = action_match.group(1).strip()
action = json.loads(action_json_str)
tool_name = action.get("tool_name")
tool_args = action.get("arguments")
if tool_name == "final_answer":
print(f"nAgent 任务完成!")
print(f"最终答案: {tool_args.get('answer')}")
return
if tool_name in available_tools:
function_to_call = available_tools[tool_name]
observation = function_to_call(**tool_args)
print(f" - 观察结果: {observation}")
messages.append({"role": "user", "content": f"Observation: {observation}"})
else:
print(f"未知工具: {tool_name}")
messages.append({"role": "user", "content": f"Observation: Error - Unknown tool '{tool_name}'."})
except Exception as e:
print(f"在步骤 {step + 1} 中发生错误: {e}")
return
print("nAgent 达到最大步骤限制，任务终止。")
async def main():
complex_task = "查询一下南京的天气，然后根据天气情况，帮我推荐一些适合的活动。"
await run_agent(complex_task)
if __name__ == "__main__":
asyncio.run(main())
MCP（Model Context Protocol，模型上下文协议）约定了AI应用如何规范地集成外部工具，实现为大模型补充上下文的目的，其本质是一个应用层协议。
Function Calling 为我们提供了调用能力，但没有一个统一规范的标准，会导致不同人开发出来的Function 有不同的传输标准，此时如果去做接入，就会有大量无意义的工作产生。同时，MCP还为我们提供了生成标准化函数Prompt的工作，我们只需要把函数文档注释写好，其就可以生成供大模型使用的提示词。
MCP中主要有三个角色，主机、服务端、客户端
环境准备
pip install mcp
服务端代码
# demo_server.py
from mcp.server.fastmcp import FastMcp
# 创建 MCP 服务器实例，命名为 "DemoServer"
mcp = FastMCP("DemoServer")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@mcp.tool()
def greet(name: str) -> str:
"""Return greeting."""
return f"Hello, {name}!"
if __name__ == "__main__":
# 使用 stdio 传输模式，方便本地调试
mcp.run(transport='stdio')
客户端代码
# demo_client.py
import asyncio
import sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 1. 配置服务器启动参数：使用当前 Python 解释器运行 server 脚本
server_params = StdioServerParameters(
command=sys.executable,
args=["demo_server.py"]
)
# 2. 建立与服务器的会话
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化连接，进行 MCP 握手
await session.initialize()
# 3. 列出服务器提供的所有可用工具
tools = await session.list_tools()
print("Available tools:", [t.name for t in tools.tools])
# 4. 调用 'add' 工具
add_result = await session.call_tool("add", arguments={"a": 3, "b": 4})
# 工具调用通常返回一个内容列表，我们提取文本内容
add_output = ''.join(content.text for content in add_result.content if hasattr(content, 'text'))
print("add(3, 4) =>", add_output)
# 5. 调用 'greet' 工具
greet_result = await session.call_tool("greet", arguments={"name": "MCP"})
greet_output = ''.join(content.text for content in greet_result.content if hasattr(content, 'text'))
print("greet('MCP') =>", greet_output)
if __name__ == "__main__":
asyncio.run(main())
本篇文章我们从一个虽然有缺陷但推理能力超强的基本LLM出发，用Prompt与之沟通，用RAG作为他的外置大脑，用Function Calling为其赋予行动能力，并通过 Agent 架构使用 LLM 最强的推理能力来指挥这一切。我们还探讨了 MCP 协议如何为这套复杂的系统提供标准的通信规范。构建出了一个强大的工作流。
但是纸上得来终觉浅，只有在实战中我们才能切身体会到上面提到的问题，才能想出如何优雅处理这些工具问题的方案。在下篇《AI应用开发实战：从零构建一个企业私有数据管家的Agent》中，我们将撸起袖子，把今天的这些理论付诸到实践中去。敬请期待！让我们一起变得更强！
文章首发gzh【破茧plan】，欢迎关注。
2025-10-09
华为鸿蒙 HarmonyOS 6.0 Developer 系统重要补丁发布，优化部分场景的使用体验
2025-10-09
三国杀移动版激活码(永久) 三国杀移动版激活码2025最新
