1. 为什么Agentic是开发范式的分水岭

先说一个我自己的经历。去年我在做一款鸿蒙应用的时候，用户提了一个很朴素的需求：我希望这个应用不只是帮我查资料，而是能根据我最近的日程、习惯和偏好，主动把需要的东西准备好。传统做法是写一堆if-else规则，再配上几个接口来回调用，但越做越觉得不对劲——规则越多，系统越脆，用户稍微换个说法，整个逻辑链条就断了。当时正好接触到HarmonyOS的AI开发工具链，再加上业界开始频繁出现Agentic这个词，我才意识到，问题不是出在代码上，而是出在开发范式上。

Agentic，直白说就是“智能体驱动”。以前我们写应用，核心是“功能”，用户自己找入口、点按钮、等结果；现在写应用，核心是“任务”，用户只需要描述目标，由AI智能体去拆解步骤、调用工具、组织信息、完成交付。这个转变看着只是交互形式变了，实际上从架构设计到数据流再到部署运维，全链路都跟着变。HarmonyOS在端侧天然有分布式能力和多设备协同优势，配合AI开发工具后，非常适合承载这种智能体形态的应用创新。拿我做的那个项目来说，最后就是用FastAPI + LangChain + LangGraph + RAG + pgvector搭了一套Agentic后端，鸿蒙端只负责交互和展示，重活全交给了智能体编排层。

这篇文章不是讲理论，而是把我从零到一搭建这套体系的完整过程拆开，包括技术选型为什么这么定、每一步的实操细节、以及我踩过的坑。如果你正准备在HarmonyOS生态里做AI应用，或者想把现有应用改造成智能体形态，这篇文章应该能帮你少走不少弯路。

1.1 从“你找功能”到“功能找你”

传统应用的信息流是单向的：用户输入指令，服务器返回结果。用户得知道“哪个功能对应哪个入口”，比如查天气要去天气页、订机票要去机票页，本质上还是人适应机器。Agentic应用则反过来，机器适应人。用户在对话框里说一句“帮我安排下周去上海出差的事”，智能体自动拆解成：查天气、订车票、找酒店、预约会议、整理行程单，然后逐个调用对应工具。这里面的关键是“拆解”和“编排”，不是简单把几个API串起来，而是让AI自己决定调用顺序和条件分支。

HarmonyOS这里有一个天然优势：端侧有大量系统能力，比如日历、邮件、地图、支付、IoT控制。这些能力如果都接到一个智能体大脑上，智能体就能真正替用户干活，而不只是陪聊。传统开发模式里，这些系统能力各自独立，开发者要写很多胶水代码才能打通；而在Agentic范式下，智能体把它们当作“可调用的工具”，按需灵活组合，整个应用就从“功能集合”变成了“任务执行者”。

1.2 HarmonyOS AI开发工具到底解决了什么问题

HarmonyOS的AI开发工具，核心目标是降低开发者构建智能体应用的门槛。几个直观感受：一是集成层面，开发工具提供了统一的能力调用框架，开发者不用自己去拼各种SDK；二是调试层面，智能体这种异步、多步执行的应用，调试比普通接口复杂得多，工具链里专门有流程追踪和状态可视化能力；三是部署层面，鸿蒙的分布式架构可以让智能体部分跑在端侧、部分跑在云端，工具会自动处理调度策略。

我实际用下来最受益的一点，是它可以让我把精力集中在“智能逻辑”上，而不是基础设施。不少人一听到Agentic应用，第一反应是“那我得从零搭一套AI推理框架”，其实不用的。HarmonyOS工具链把周边琐事处理掉之后，核心要做的事情就三件：定义智能体的行为边界、设计工具调用流程、做好知识库的检索质量。这三件事恰恰是最能体现业务价值的地方。

2. HarmonyOS AI应用的技术栈选型分析

标题里提到的那串技术栈——FastAPI + LangChain + LangGraph + RAG + pgvector，不是随便拼的，每个组件在我这套系统里都有明确的位置。接下来逐个拆解，顺便说清楚为什么是它们，而不是别的。

2.1 工具选型：为什么是FastAPI + LangChain + LangGraph

先看后端框架。鸿蒙端的应用是ArkTS写的，但AI智能体的业务逻辑放在云端更灵活，所以需要一个后端服务来承载。选FastAPI的原因很实际：它是Python生态里最顺手的异步Web框架，性能好、类型提示完善、自动生成OpenAPI文档，这对AI应用来说非常省事。因为LangChain生态是Python主导的，选FastAPI可以直接用Python完成“接口层 + 智能体逻辑层”的融合，不需要引入跨语言桥接。如果你非要选别的语言，可能就得面对“SDK能力不齐、工具支持不完整”的局面。

再来看LangChain和LangGraph的分工。很多初学者把LangChain当成了整个Agent框架本身，实际上LangChain更准确的角色是“组件库”，提供模型接入、Prompt管理、输出解析、工具封装等基础零件。真正把Agent行为编排起来的是LangGraph，它把Agent的运行过程抽象成一张有向图：节点是“任务步骤”，边是“状态流转路径”，每一步都有明确的输入输出。这样设计的好处是，你可以清楚地看到智能体走到了哪一步、为什么走这一步、出问题时在哪一步出问题。相比直接把逻辑写死在一大段代码里，图结构的可观测性和可维护性都好太多了。

2.2 RAG与pgvector在知识检索中的角色

Agent光会调用工具还不够，很多时候它需要回答基于私有知识的问题。比如企业内部的规范文档、产品的使用手册、历史工单的处理经验，这些内容不可能全塞进大模型的上下文里，所以要用RAG（检索增强生成）。RAG的流程大致是：先把文档切块、向量化、存进向量数据库；用户提问时，把问题也向量化，去库里检索最相关的内容块；最后把检索结果和用户问题一起交给大模型生成答案。

pgvector在这个体系里负责的是“存向量、查向量”。有人会问，专门的向量数据库那么多，为什么选pgvector？我的答案是：省心。团队里已经有PostgreSQL在跑业务数据，直接在同一个实例里扩展出向量检索能力，不用单独维护一套新的存储系统。而且pgvector支持HNSW索引，数据量在百万级以下时，召回速度和精度都不错，对绝大多数HarmonyOS生态的应用场景够用了。它不是性能最强的选项，却是工程复杂度最低、团队上手成本最小的选项。

2.3 后端与鸿蒙端的协作边界

还有一个我经常被问到的问题：智能体逻辑到底应该放在鸿蒙端还是云端？我的建议是分层。轻量级、延迟敏感的部分，比如一句话指令的意图判断，放在端侧，让鸿蒙的端侧AI能力直接处理；需要大量知识检索、多步规划、或者要访问大模型的部分，放到云端后端。这样分工的理由是：端侧资源有限，但响应快、隐私好；云侧算力充足、生态丰富，但多一跳网络延迟。HarmonyOS的分布式架构天然支持这种混合部署，关键是要在架构设计阶段就把边界划清楚，否则后期改起来很痛苦。

3. 搭建一个鸿蒙侧AI智能体的完整过程

理论说了不少，现在进入实操。这部分的最终目标是：鸿蒙应用里发一句话，云端智能体拆解任务，检索知识库，调用工具，回传结果。我按步骤写，你跟着做就能跑通。

3.1 环境准备与项目结构

先说环境。后端部分需要Python 3.10以上，PostgreSQL 14以上并安装pgvector扩展。以下是核心依赖，我用requirements.txt管理，你直接复制安装就行：

fastapi==0.110.0
uvicorn[standard]==0.29.0
langchain==0.1.16
langchain-openai==0.1.7
langgraph==0.0.62
pgvector==0.2.5
psycopg2-binary==2.9.9
sentence-transformers==2.6.1
pydantic==2.6.4

安装命令：

pip install -r requirements.txt

PostgreSQL这边要启用pgvector扩展。连上数据库后执行：

CREATE EXTENSION IF NOT EXISTS vector;

如果提示没有这个扩展，说明你的PostgreSQL版本或安装方式里没带pgvector，需要先编译安装，Debian系的系统也可以试试 apt install postgresql-14-pgvector ，版本号按你自己的PostgreSQL版本调整。

项目结构我习惯这么组织，逻辑清晰，后期扩功能也方便：

harmony_agent/
├── app.py                 # FastAPI入口
├── agent/
│   ├── graph.py           # LangGraph状态图定义
│   ├── nodes.py           # 各节点逻辑
│   └── tools.py           # 工具封装
├── rag/
│   ├── embedder.py        # 向量化封装
│   ├── retriever.py       # 检索器
│   └── vector_store.py    # pgvector连接与操作
├── config.py              # 全局配置
└── data/
    └── knowledge_base/    # 知识库原始文件

3.2 后端核心模块实现

先从FastAPI入口写起。这个入口做的事情只有三件：接收鸿蒙端发来的用户请求、调用智能体编排层处理、返回结果。我加了一层请求ID追踪，方便排查链路问题。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent.graph import run_agent

app = FastAPI(title="Harmony Agent API")

class AgentRequest(BaseModel):
    user_id: str
    message: str
    session_id: str = "default"

class AgentResponse(BaseModel):
    answer: str
    trace_id: str

@app.post("/api/agent", response_model=AgentResponse)
async def handle_agent_request(req: AgentRequest):
    trace_id = f"{req.user_id}-{req.session_id}"
    try:
        result = await run_agent(
            user_id=req.user_id,
            message=req.message,
            session_id=req.session_id
        )
        return AgentResponse(answer=result, trace_id=trace_id)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent execution failed: {str(e)}")

这里有个细节： run_agent 是async函数，因为LangGraph的执行过程里包含LLM调用、HTTP工具调用、向量库查询，全是IO密集操作，异步能有效提升并发吞吐。我在实测中，同步版本单机同时处理10个请求就开始有显著延迟，改异步之后20个请求依然稳定。虽然FastAPI本身支持同步函数自动转线程池，但显式async能让调度更可控。

3.3 LangGraph状态图：核心编排逻辑

LangGraph是这套体系里最需要花心思的地方。我用它定义了一个四级状态机：意图识别、任务规划、工具执行、结果生成。每一级是图上的一个节点，节点之间通过状态对象传递数据，这样每一步都可以单独测试和观察。

from typing import TypedDict, List
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    user_message: str
    intent: str
    plan: List[str]
    tool_results: dict
    final_answer: str

def intent_node(state: AgentState) -> AgentState:
    # 用LLM判断意图，输出结构化结果
    prompt = f"判断用户意图，输出为：查询知识库、调用工具、闲聊。用户输入：{state['user_message']}"
    state["intent"] = call_llm(prompt).strip()
    return state

def plan_node(state: AgentState) -> AgentState:
    # 根据意图生成任务步骤列表
    state["plan"] = generate_plan(state["user_message"], state["intent"])
    return state

def execute_node(state: AgentState) -> AgentState:
    # 遍历plan，逐个执行工具调用或知识库检索
    results = {}
    for step in state["plan"]:
        if step.startswith("rag:"):
            results[step] = retrieve_from_kb(step[4:])
        elif step.startswith("tool:"):
            results[step] = call_tool(step[5:], state["user_message"])
    state["tool_results"] = results
    return state

def generate_node(state: AgentState) -> AgentState:
    state["final_answer"] = generate_final_answer(
        state["user_message"], state["tool_results"]
    )
    return state

graph = StateGraph(AgentState)
graph.add_node("intent", intent_node)
graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)
graph.add_node("generate", generate_node)

graph.set_entry_point("intent")
graph.add_edge("intent", "plan")
graph.add_edge("plan", "execute")
graph.add_edge("execute", "generate")
graph.add_edge("generate", END)

我踩过的一个坑：最开始我把所有节点逻辑都写在一个函数里，结果一旦中间某步出错，整个会话就断了，日志里只能看到一行异常，根本定位不了是哪一步的问题。改成图结构之后，每个节点可以单独打印状态、单独设置超时、单独重试，排查效率完全不一样。

3.4 RAG召回流程与pgvector配置

RAG这块如果只做“文档切块 -> 塞向量库 -> 向量检索”，效果其实一般。我实际跑下来，有两个环节对召回质量的提升最明显：一是切块策略，二是查询改写。

切块策略方面，我强烈建议不要用固定的“按字符数硬切”，而要按语义边界来切。比如文档里有标题、段落、列表，就应该按这些结构断开。我用的是递归字符切分器，但把分隔符优先级调成了：段落标记 > 换行符 > 句号 > 分号 > 逗号。这样切出来的块在语义上相对完整，检索时更精准。块大小我试过从256到1024，最终选512，单块内容适中，既能覆盖完整语义，又不至于让上下文窗口被无关内容占满。

查询改写方面，用户的问题往往很口语化，直接拿去向量检索效果差。比如用户问“咱们年假政策是啥”，直接检索“年假政策是啥”可能搜不到文档里的“休假管理细则”。我会先用一个改写Prompt，把口语问题转成更接近文档表述的查询词，再去做向量化检索。这一步的收益很大，实测top-5命中率提升了两成左右。

pgvector建表与索引配置如下：

CREATE TABLE IF NOT EXISTS kb_docs (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    source VARCHAR(255),
    embedding vector(768)
);

CREATE INDEX ON kb_docs USING hnsw (embedding vector_cosine_ops);

这里有两个参数值得注意。第一是 dimension，768 对应我用的 embedding 模型输出维度，你用其他模型就要改成对应的维度，比如 OpenAI 的 text-embedding-3-small 是 1536 维，BGE 系列常见的是 768 维或 1024 维。第二是索引类型 HNSW，它是近似最近邻搜索算法，召回速度远快于暴力搜索，但对内存要求略高。数据量十万级以内建议 m=16, ef_construction=64 ，多了容易内存吃紧。检索时我用余弦距离，因为语义相似度用余弦比欧氏更稳定。

检索核心代码如下：

from pgvector.psycopg2 import register_vector
import psycopg2

def search_kb(query: str, top_k: int = 5):
    conn = psycopg2.connect(CONN_STR)
    register_vector(conn)
    cur = conn.cursor()
    
    query_vector = embed_text(query)
    cur.execute("""
        SELECT content, 1 - (embedding <=> %s) AS similarity
        FROM kb_docs
        ORDER BY embedding <=> %s
        LIMIT %s
    """, (query_vector, query_vector, top_k))
    
    results = cur.fetchall()
    cur.close()
    conn.close()
    return results

注意 <=> 是余弦距离算子，用 1 - 距离 转成相似度分数，这样分数越高代表越相关。下面这部分容易被忽略：检索出来的内容块虽然不是越多越好，但当单块内容确实不够时，我会做一个简单的重排——不是直接按相似度取前五，而是用MMR（最大边际相关性）去重，避免五条结果全在讲同一件事。pgvector本身没有内置MMR，我是在拿到候选集后在代码里实现的。这一步对答案质量的影响很大，尤其是知识库里重复内容多的时候。

3.5 鸿蒙端接入与联调

后端跑通之后，鸿蒙端的接入反而简单。用ArkTS的HTTP模块发请求即可：

import http from '@ohos.net.http';

async function sendAgentRequest(message: string) {
  let httpRequest = http.createHttp();
  let response = await httpRequest.request(
    'https://your-server.example.com/api/agent',
    {
      method: http.RequestMethod.POST,
      header: { 'Content-Type': 'application/json' },
      extraData: JSON.stringify({
        user_id: 'user_123',
        message: message,
        session_id: 'session_456'
      })
    }
  );
  return JSON.parse(response.result as string).answer;
}

这里需要留意的是，鸿蒙应用请求云端接口时，如果目标是公网域名，必须配置网络安全信任关系，开发阶段可以在 module.json5 里把网络权限开一下，同时把目标域名加入 network_security_config 的白名单。不配的话会直接报 SSL 错误，我第一次联调就被这个卡了半小时。另外，如果后端有鉴权，建议在请求头里带上 token，最好是短时有效的动态 token，不要写死。这不仅是安全规范问题，也关系到后续上架审核能否通过。

4. 开发中的常见问题与排查技巧

这部分我整理了一张问题表，都是我在实际开发中遇过的，不是网上复制来的。按“现象 -> 原因 -> 解决”的顺序给你参考。

现象	可能原因	解决方案
向量检索返回完全无关结果	embedding模型维度和pgvector表维度不一致，或者切块粒度太粗	检查 vector(768) 与模型输出维度是否一致；改用语义边界切块
Agent执行到中途无响应	某个工具节点阻塞，可能是第三方接口超时	给每个工具节点加超时控制，超时自动进入兜底节点
鸿蒙端请求返回SSL错误	域名未加入网络安全白名单	在 network_security_config 中配置回溯，开发期可关闭校验但上架前必须恢复
并发一高，后端延迟骤增	同步阻塞型工具调用占用了线程池	把同步工具改为async调用，或创建独立线程池
LangGraph状态错乱	在多轮会话中复用了单轮State对象	每次会话创建独立State实例，不跨请求共享
答案“一本正经胡说”	RAG检索到的上下文与问题无关，或检索为空时仍强制生成	设置检索置信度阈值，低于阈值时让Agent明确说“无法回答”

4.1 向量检索效果差的排查思路

这类问题遇到最多次。先别急着调模型，按这个顺序排查：第一步，查数据质量。知识库源文档里有大量扫描版PDF、格式混乱的表格，切出来的块是乱码，神仙模型也救不回来。解决方法是清洗文档，表格转成结构化文本之后再入库。第二步，查切块策略。对比同一文档在不同切块参数下的召回效果，你可以写一个最小验证集，挑二三十个典型问题，人工标注每道题的正确答案对应哪段文档，然后用不同参数跑检索，统计top-5命中率。我自己的参数就是这么调出来的，而不是靠感觉。第三步才去考虑换更强的embedding模型或加重排模型。

4.2 LangGraph状态机常见的坑

用LangGraph容易踩的坑：一是节点函数里意外修改了共享状态对象，导致并发请求互相污染。解决的硬规矩是：节点函数体内部生成的临时数据，一律用局部变量，只在最后return时更新state里需要传递的字段。二是条件边的写法。LangGraph的条件边是“返回下一节点名称”的函数，如果你返回了不存在的节点名，它会静默走默认路径，不会报错。排查方法是在条件函数里打日志，确认返回路径符合预期。三是子图的使用。当流程复杂到超过十个节点，单图会非常难维护，可以把“工具调用组”单独抽成一个子图，主图里只保留一个节点指向子图入口。

我在项目里专门加了一层“人工确认”节点：凡是涉及发送消息、删除数据这类敏感操作的步骤，Agent生成结果后不直接执行，而是先在图上停留，返回一个“确认清单”给鸿蒙端用户，用户点了确认才继续执行。这个设计在日志里追踪起来也方便，因为能看到Agent在“等待确认”节点停留了多久，用户最终选择了接受还是拒绝。

4.3 给新手的快速调试建议

本地调试时，别每次都把鸿蒙应用跑起来测，那太慢了。我的习惯是：先用curl直接打FastAPI接口，验证Agent逻辑；再用Python脚本模拟多轮对话，验证会话状态；最后才回到鸿蒙端连真机验证。另外强烈建议开LangSmith或类似的追踪工具，它能把你每一次Agent运行的中间步骤可视化出来，哪个节点耗时长、哪步Prompt导致结果不对，一眼就能看到。效率比盯着日志硬猜高一个量级。

5. 一点个人心得

做这个项目最大的体会是：Agentic开发不是“多了一个新功能”，而是把之前“预先把所有路径写死”的思路整个反转了。传统应用像铁路，轨道铺到哪，火车到哪；Agentic应用更像出租车平台，用户报个目的地，系统动态规划路线、选择车型、处理拼单，一切都围绕“任务”而不是“班次”。这套思路放到HarmonyOS生态里，潜力很大，因为鸿蒙的多设备协同能力给了Agent更多的执行空间——手机、平板、手表、智能家居都可以成为Agent调用的工具。但前提是，你的后端编排层要扛得住这种灵活度，这也是我为什么在LangGraph上花了大精力。

最后分享一个我觉得特别重要的小技巧：AI应用的日志，一定要结构化。不要只记“调用了某某模型”，要把输入、输出、耗时、消耗的Token数、用到的工具、检索到的文档ID全都记下来。项目上线之后，用户说“答案不对”，你翻日志能快速还原现场，而不是跟用户来回拉扯半天。这个习惯救了我很多次，推荐所有做Agentic开发的同学都养成。

如果你正在折腾HarmonyOS同类型项目，建议从小场景切入，比如做一个内部知识问答助手，或者一个自动整理日程的执行助手，先把RAG和LangGraph这两个核心跑通，再逐步加工具调用和多轮记忆。等你的Agent能稳定完成10个步骤以内的任务，再去做更复杂的编排，这样体验会平滑很多。