1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from llama_cloud_services import LlamaParse

parser = LlamaParse(
    api_key="your_api_key",
    result_type="markdown",
    num_workers=3,
    language="ch_sim",  # 中文简体
)

file_extractor = {".pdf": parser}
documents_cloud = SimpleDirectoryReader(
    "./data", file_extractor=file_extractor
).load_data()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

def chunk_text(text, max_chunk_size=300):
    lines = text.split('\n')
    chunks = []
    current_chunk = []
    current_size = 0
    in_table = False
    table_content = []

    for line in lines:
        # 检测表格开始（含有 | 字符的行）
        if '|' in line and not in_table:
            in_table = True
            if current_chunk:
                chunks.append('\n'.join(current_chunk))
                current_chunk = []
                current_size = 0
            table_content.append(line)
        elif in_table:
            table_content.append(line)
            if not line.strip():  # 空行表示表格结束
                in_table = False
                chunks.append('\n'.join(table_content))
                table_content = []
        else:
            line_length = len(line)
            if current_size + line_length > max_chunk_size and current_chunk:
                chunks.append('\n'.join(current_chunk))
                current_chunk = []
                current_size = 0
            current_chunk.append(line)
            current_size += line_length
    # 处理剩余内容...
    return chunks

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from milvus_model.hybrid import BGEM3EmbeddingFunction

# 加载 BGE-M3 模型
ef = BGEM3EmbeddingFunction(
    model_name_or_path="BAAI/bge-m3",
    use_fp16=False,
    device="cpu"
)

# 生成嵌入向量
chunks_embeddings = ef(chunks)
# chunks_embeddings["dense"]  -> 稠密向量列表
# chunks_embeddings["sparse"] -> 稀疏向量矩阵

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from pymilvus import (
    connections, utility, FieldSchema,
    CollectionSchema, DataType, Collection,
)

connections.connect(uri="./milvus.db")

fields = [
    FieldSchema(name="pk", dtype=DataType.VARCHAR,
                is_primary=True, auto_id=True, max_length=100),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=512),
    FieldSchema(name="sparse_vector", dtype=DataType.SPARSE_FLOAT_VECTOR),
    FieldSchema(name="dense_vector", dtype=DataType.FLOAT_VECTOR,
                dim=dense_dim),
]

schema = CollectionSchema(fields)
col = Collection("hybrid_demo", schema, consistency_level="Strong")

# 分别为两种向量创建索引
col.create_index("sparse_vector",
                 {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"})
col.create_index("dense_vector",
                 {"index_type": "AUTOINDEX", "metric_type": "IP"})
col.load()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from pymilvus import AnnSearchRequest, WeightedRanker

def hybrid_search(col, query_dense_embedding, query_sparse_embedding,
                  sparse_weight=1.0, dense_weight=1.0, limit=10):
    # 构建稠密向量搜索请求
    dense_req = AnnSearchRequest(
        [query_dense_embedding], "dense_vector",
        {"metric_type": "IP", "params": {}}, limit=limit
    )
    # 构建稀疏向量搜索请求
    sparse_req = AnnSearchRequest(
        [query_sparse_embedding], "sparse_vector",
        {"metric_type": "IP", "params": {}}, limit=limit
    )
    # 使用 WeightedRanker 融合两路结果
    rerank = WeightedRanker(sparse_weight, dense_weight)
    res = col.hybrid_search(
        [sparse_req, dense_req], rerank=rerank,
        limit=limit, output_fields=["text"]
    )[0]
    return [hit.get("text") for hit in res]

1
2
3
4
5
6
7

hybrid_results = hybrid_search(
    col,
    query_embeddings["dense"][0],
    query_embeddings["sparse"]._getrow(0),
    sparse_weight=0.7,  # 稀疏权重
    dense_weight=1.0,   # 稠密权重
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def rag(query):
    # 1. 检索
    formatted_references = format_list_with_markers(retrieval(query))

    # 2. 构造带引用的 Prompt
    prompt = f"""
    你是一个智能助手，负责根据用户的问题和提供的参考内容生成回答。
    1. 回答必须基于提供的参考内容。
    2. 每一块内容必须标注引用来源，格式为：[引用编号]。
    3. 如果没有参考内容，根据你自己的知识进行回答。

    参考内容：
    {formatted_references}

    用户问题：{query}
    """

    # 3. 调用 LLM 生成回答
    completion = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return formatted_references, completion.choices[0].message.content

1
2

# 在 websearch 模块的 fetch_web_content.py 中
# Serper API 返回搜索结果列表，包含标题、链接、摘要

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class WebScraper:
    def __init__(self, user_agent='macOS'):
        self.headers = self._get_headers(user_agent)

    def get_webpage_html(self, url):
        if url.endswith(".pdf"):
            return requests.Response()  # 跳过 PDF
        try:
            response = requests.get(url, headers=self.headers, timeout=8)
            response.encoding = "utf-8"
        except requests.exceptions.Timeout:
            return requests.Response()
        return response

    def extract_main_content(self, html_soup, rule=0):
        main_content = []
        # 只提取 h1-h6 和 p 标签的内容
        tag_rule = re.compile("^(h[1-6]|p)")
        for tag in html_soup.find_all(tag_rule):
            tag_text = tag.get_text().strip()
            if tag_text and len(tag_text.split()) > 10:
                main_content.append(tag_text)
        return "\n".join(main_content).strip()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

class EmbeddingRetriever:
    TOP_K = 20

    def retrievel_embeddings(self, contents_list, link_list, query):
        metadatas = [{'url': link} for link in link_list]
        texts = self.text_splitter.create_documents(
            contents_list, metadatas=metadatas
        )
        # 使用 SentenceTransformer 进行嵌入
        embedding = SentenceTransformerEmbeddings(
            model_name='sentence-transformers/all-MiniLM-L6-v2'
        )
        # 使用 Chroma 向量数据库进行相似度检索
        db = Chroma.from_documents(documents=texts, embedding=embedding)
        retriever = db.as_retriever(
            search_kwargs={"k": min(self.TOP_K, len(texts))}
        )
        return retriever.get_relevant_documents(query)

1
2

用户查询 -> Serper API 搜索 -> 多线程爬取网页 -> 提取正文内容
         -> 向量化 + Chroma 检索 -> 返回最相关的片段

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

def agent_plan(query):
    prompt = '''
    你是一个专业的汽车销售助手的规划模块。你的任务是：
    1. 分析用户的查询:{0}
    2. 决定使用哪个工具（本地文档搜索或网络搜索）
    3. 将原始查询拆解或延伸为1-2个相关问题

    ## 工具选择规则
    - 涉及星辰电动ES9的具体信息时 -> 本地文档搜索
    - 涉及竞品对比、最新动态、实时数据时 -> 网络搜索

    ## 输出格式（JSON）
    [
      {{
        "action_name": "工具名称",
        "prompts": ["原始查询", "拆解问题1", "拆解问题2"]
      }}
    ]
    '''.format(query)
    result = middle_json_model(prompt)
    return json.loads(extract_json_content(result))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def process_actions(actions):
    memory = []
    for action in actions:
        action_name = action['action_name']
        prompt = action['prompt']

        try:
            if action_name == '本地文档搜索':
                result = rag(prompt)
            elif action_name == '网络搜索':
                result = web_search_answer(prompt)
            else:
                result = f"未知的动作类型: {action_name}"

            memory.append({"提问": prompt, "结果": result})
        except Exception:
            continue

    return memory

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def final_answer(memory_global, user_query):
    final_prompt = f'''
    你是一个星辰电动ES9的智能销售助手。
    基于提供的参考内容进行回答，
    用有打动力的销售语言输出，突出星辰电动的优势。

    参考内容：{memory_global}
    用户问题：{user_query}
    '''
    completion = client.chat.completions.create(
        model="deepseek-r1",
        messages=[{"role": "user", "content": final_prompt}],
        stream=True,
    )
    # 流式输出思考过程和最终回答...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

def reflection(user_query, memory_global):
    prompt = '''
    你是一个专业的汽车销售助手的规划模块。
    1. 分析用户的查询:{0}
    2. 基于已有的信息，是否还需要延伸再进行查询

    已有的信息: {1}

    至多再扩展不超过3个查询，
    如果需要扩展则输出JSON格式，如果不需要则返回None
    '''.format(user_query, memory_global)
    result = middle_json_model(prompt)
    return json.loads(extract_json_content(result))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

def agent_reflection(user_query):
    memory_global = []
    # 第一轮：规划 + 执行
    action_tool = agent_plan(user_query)
    actions = adjust_format(action_tool) if action_tool else []
    if actions:
        memory_new = process_actions(actions)
        memory_global.extend(memory_new[1:])

    # 反思：评估信息是否充足，决定是否需要补充
    action_reflect = reflection(user_query, memory_global)
    if action_reflect:
        print("回顾内容，进行反思...")
        memory_new = process_actions(action_reflect)
        memory_global.extend(memory_new)

    # 最终回答
    final_answer(memory_global, user_query)

1
2
3
4
5
6
7
8
9

┌──────────────┐    MCP协议    ┌──────────────┐
│   MCP Host   │◄────────────►│  MCP Server  │
│ (如 Claude   │              │ (工具/服务)    │
│  Desktop)    │              │              │
│  ┌────────┐  │              └──────────────┘
│  │MCP     │  │    MCP协议    ┌──────────────┐
│  │Client  │◄─┼────────────►│  MCP Server  │
│  └────────┘  │              │ (另一个工具)   │
└──────────────┘              └──────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("weather", log_level="ERROR")

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """向天气 API 发送请求"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)
    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."
    alerts = [format_alert(f) for f in data["features"]]
    return "\n---\n".join(alerts)

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)
    if not points_data:
        return "Unable to fetch forecast data for this location."
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)
    if not forecast_data:
        return "Unable to fetch detailed forecast."
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']}{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)
    return "\n---\n".join(forecasts)

if __name__ == "__main__":
    mcp.run(transport='stdio')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

from mcp.server import FastMCP
from mcp.types import TextContent

mcp = FastMCP("Search Tools Server")

class SearchTool:
    def __init__(self):
        self.api_key = os.getenv("SERPER_API_KEY")
        self.base_url = "https://google.serper.dev/search"

    async def search(self, query: str, num_results: int = 10):
        headers = {
            "X-API-KEY": self.api_key,
            "Content-Type": "application/json"
        }
        payload = {"q": query, "num": num_results,
                   "gl": "cn", "hl": "zh-cn"}
        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.base_url, headers=headers,
                json=payload, timeout=30.0
            )
            return response.json()

search_tool = SearchTool()

@mcp.tool()
async def web_search(query: str, num_results: int = 10):
    """执行网络搜索
    Args:
        query: 搜索查询词
        num_results: 返回结果数量，默认10条
    """
    result = await search_tool.search(query, num_results)
    # 格式化并返回结果...

@mcp.tool()
async def news_search(query: str, num_results: int = 10):
    """执行新闻搜索"""
    # ...

@mcp.tool()
async def weather_search(city: str):
    """查询天气信息"""
    # ...

if __name__ == "__main__":
    mcp.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

class LLMWithMCP:
    def __init__(self):
        self.client = OpenAI(
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
        )
        self.tools = []           # OpenAI 格式的工具定义
        self.mcp_tools_info = []  # MCP 工具详细信息

    async def initialize_mcp(self):
        """连接 MCP Server 并自动发现工具"""
        server_params = StdioServerParameters(
            command="python",
            args=["mcp_server.py"],
            env=None
        )
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()

                # 自动发现工具
                tools_result = await session.list_tools()
                for tool in tools_result.tools:
                    # 将 MCP 工具定义转换为 OpenAI 格式
                    openai_tool = {
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "description": tool.description,
                            "parameters": tool.inputSchema
                        }
                    }
                    self.tools.append(openai_tool)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def build_system_prompt(self) -> str:
    """根据可用工具动态构建系统提示词"""
    tools_desc = []
    for i, tool in enumerate(self.mcp_tools_info, 1):
        tools_desc.append(f"{i}. {tool['name']} - {tool['description']}")
    tools_text = "\n".join(tools_desc)

    return f"""你是一个智能助手，具有以下工具能力：

{tools_text}

必须使用工具的情况：
- 用户明确要求"搜索"、"查找"某个话题
- 用户询问实时数据：股价、天气、新闻等
- 用户询问2023年之后的信息

不要使用工具的情况：
- 解释基本概念
- 回答常识性问题
- 提供教程或指导"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

async def chat_with_tools(self, user_message: str) -> str:
    messages = [
        {"role": "system", "content": self.build_system_prompt()},
        {"role": "user", "content": user_message}
    ]

    # 第一次调用 LLM（传入工具定义）
    response = self.client.chat.completions.create(
        model=self.model_name,
        messages=messages,
        tools=self.tools,
    )
    assistant_message = response.choices[0].message

    # 如果 LLM 决定调用工具
    if assistant_message.tool_calls:
        messages.append(assistant_message)  # 记录 LLM 的工具调用决策

        for tool_call in assistant_message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            # 通过 MCP 调用实际工具
            tool_result = await self.call_mcp_tool(
                function_name, function_args
            )

            # 将工具结果回传给 LLM
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": tool_result
            })

        # 第二次调用 LLM（基于工具结果生成最终回答）
        final_response = self.client.chat.completions.create(
            model=self.model_name,
            messages=messages,
            # 注意：此处不传 tools 参数，避免再次触发工具调用
        )
        return final_response.choices[0].message.content
    else:
        return assistant_message.content

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# agent.py 中的核心函数

def final_answer(user_query):
    """完整的 Agent 流程（Generator，支持 SSE 流式输出）"""

    # 1. 规划
    action_tool = agent_plan(user_query)
    actions = adjust_format(action_tool) if action_tool else []

    # 2. 向前端推送当前执行状态
    for action in actions:
        message = {
            "role": "agent",
            "content": f'正在执行{action["action_name"]}: "{action["prompt"]}"'
        }
        yield f"event: message\ndata: {json.dumps(message)}\n\n"

    # 3. 执行检索并去重
    memory_new = process_actions(actions)
    memory_global = list(memory_new)[1:]

    # 4. Reflexion 反思
    action_reflect = reflection(user_query, memory_global)
    if action_reflect:
        memory_new = process_actions(actions)
        memory_global.extend(memory_new)

    # 5. 流式输出最终回答（使用 DeepSeek-R1）
    completion = client.chat.completions.create(
        model="deepseek-r1",
        messages=[{"role": "user", "content": final_prompt}],
        stream=True,
    )
    for chunk in completion:
        if chunk.choices[0].finish_reason == "stop":
            yield "event: end\ndata: [DONE]\n\n"
            break
        delta = chunk.choices[0].delta
        if delta.content:
            yield f"event: message\ndata: {json.dumps(...)}\n\n"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

AI Agent 与 MCP 协议
├── 1. AI Agent 基础
│   ├── 定义：以 LLM 为大脑的自主系统
│   ├── 五大模块
│   │   ├── Planning（计划）
│   │   ├── Memory（记忆）
│   │   ├── Tools（工具）
│   │   ├── Executor（执行）
│   │   └── Output（输出）
│   └── vs 传统对话式 AI
│
├── 2. RAG 组件
│   ├── 文档解析（LlamaParse）
│   ├── 文本切块（chunk_text）
│   ├── 向量化（BGE-M3）
│   │   ├── 稠密向量（语义匹配）
│   │   └── 稀疏向量（关键词匹配）
│   ├── 向量数据库（Milvus）
│   └── 混合检索（WeightedRanker）
│
├── 3. Web 搜索模块
│   ├── Serper API 搜索
│   ├── 多线程网页爬取（WebScraper）
│   ├── 内容提取（h/p 标签过滤）
│   └── 相似度重排序（Chroma + SentenceTransformer）
│
├── 4. Agent 决策流程
│   ├── Planning：工具选择 + 查询拆解
│   ├── Execution：RAG / Web Search
│   └── Output：综合生成
│
├── 5. 高级框架
│   ├── ReAct（推理 + 行动交织）
│   └── Reflexion（自我反思）
│       ├── Actor（行动者）
│       ├── Evaluator（评估者）
│       └── Self-Reflection（自我反思）
│
├── 6. MCP 协议
│   ├── 核心概念
│   │   ├── Host（宿主应用）
│   │   ├── Client（客户端）
│   │   └── Server（服务端）
│   ├── MCP Server 编写
│   │   ├── FastMCP 初始化
│   │   ├── @mcp.tool() 装饰器
│   │   └── transport='stdio' 启动
│   ├── MCP Client 实现
│   │   ├── 自动工具发现（list_tools）
│   │   ├── 动态 System Prompt
│   │   └── 工具调用流程（两次 LLM 调用）
│   └── MCP 生态
│       ├── Client 工具（Claude Desktop, Cherry Studio）
│       └── Server 平台（mcp.so, mcpmarket.com）
│
└── 7. 完整项目：SalesPilot
    ├── Agent 流程编排
    ├── SSE 流式输出
    ├── Memory 去重
    └── DeepSeek-R1 推理输出