1. 从问答到执行：RAG系统的业务能力升级之路
作为长期从事AI工程化的开发者，我深刻体会到单纯的知识问答系统在实际业务场景中的局限性。去年在为某金融科技公司实施RAG系统时，技术团队最初欢呼雀跃于其文档检索能力，但两周后业务部门的反馈却给我们泼了冷水："它能准确回答我们的问题，但我们需要的是能直接完成风险报告生成的工具，而不是又一个需要人工整理的问答机。"
这个案例揭示了RAG系统的关键瓶颈：知道不等于会做。传统RAG就像个博学的图书管理员，能快速找到相关资料，但业务人员需要的是能按标准流程完成特定任务的专业助手。本文将分享如何通过Agent和Skill架构，将RAG升级为真正的业务执行引擎。
2. 架构设计：三层能力模型解析
2.1 基础层：RAG作为知识引擎
RAG核心价值在于将非结构化文档转化为可检索的知识。在我们的实践中，Python技术栈通常这样构建：
python复制from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
docs = text_splitter.split_documents(documents)
vectorstore = FAISS.from_documents(docs, embedding_model)
关键参数选择依据：
- chunk_size=1000：平衡检索精度与上下文完整性
- overlap=200：确保关键概念不被分割
- FAISS：轻量级且支持增量更新
2.2 能力层：Skill的标准化封装
Skill的本质是将业务SOP（标准操作流程）转化为机器可执行的规范。以代码审查为例，我们定义的SKILL.md包含：
markdown复制---
name: "code-review"
input_schema:
code: str
language: Literal["python","go","java"]
output_schema:
security_issues: List[Dict]
performance_issues: List[Dict]
style_violations: List[Dict]
overall_score: float
---
这种声明式定义带来三个优势：
- 人机可读：开发者与业务方对需求理解一致
- 接口明确：输入输出结构化，便于系统集成
- 可复用：相同模式可快速创建合同审查、运维诊断等Skill
2.3 调度层：轻量Agent的实现策略
不同于复杂的Agent框架，我们采用"路由+验证"的最小化设计：
python复制class CodeReviewAgent:
def __init__(self):
self.skill = CodeReviewSkill()
self.validator = PydanticValidator()
async def execute(self, request: dict):
if not self.validator.validate_input(request):
raise InvalidRequestError()
try:
result = await self.skill.run(
code=request["code"],
language=request.get("language","python")
)
return self.validator.format_output(result)
except Exception as e:
log_error(e)
raise ExecutionError()
这种设计在保证核心功能的前提下，将复杂度控制在200行代码以内，更易于维护和扩展。
3. 代码审查Skill的完整实现
3.1 知识库构建最佳实践
针对代码审查场景，我们构建了多维度知识库：
code复制/kb-code-style
├── security
│ ├── owasp_top_10.md
│ └── python_safety.md
├── performance
│ ├── database_optimization.md
│ └── python_profiling.md
└── style_guides
├── pep8.md
└── google_java_guide.md
索引构建时采用混合分块策略：
- 规范类文档：按章节分块（保持完整性）
- 案例类文档：按问题类型分块（便于精准检索）
3.2 审查逻辑的核心实现
review.py的关键组件：
python复制def analyze_code(code: str, language: str) -> dict:
# 知识检索
knowledge = retrieve_related_knowledge(code, language)
# 多维度分析
security = SecurityAnalyzer(knowledge).check(code)
performance = PerformanceAnalyzer(knowledge).check(code)
style = StyleChecker(knowledge).check(code)
# 综合评估
return generate_report(
security=security,
performance=performance,
style=style,
template=load_template("code_review.md")
)
其中Prompt工程的关键点：
- 提供审查维度明确定义
- 要求引用具体知识条目
- 输出严格遵循模板
3.3 性能优化实战技巧
在处理大代码库时，我们采用分级审查策略：
-
预过滤阶段（快速模型如gpt-3.5-turbo）：
- 识别代码关键段落
- 确定主要审查方向
-
深度分析阶段（强模型如gpt-4）：
- 聚焦关键问题点
- 结合更多上下文
这使平均处理时间从28秒降至9秒，成本降低60%。
4. 生产环境部署方案
4.1 服务化架构设计
code复制 +-----------------+
| Client |
+--------+--------+
|
+--------v--------+
| API Gateway |
+--------+--------+
|
+---------------+---------------+
| |
+---------v---------+ +-----------v-----------+
| Code Review API | | Other Skill APIs |
+---------+---------+ +-----------+-----------+
| |
+---------v---------+ +-----------v-----------+
| Skill Executor | | Knowledge Service |
+-------------------+ +-----------------------+
4.2 关键配置示例
FastAPI应用的典型配置：
python复制app = FastAPI(
title="Skill Service",
version="1.0.0",
middleware=[
Middleware(TrustedHostMiddleware),
Middleware(GZipMiddleware)
]
)
@app.post("/review")
async def code_review(
request: CodeReviewRequest,
background_tasks: BackgroundTasks
):
task = background_tasks.add_task(
execute_review,
request.code,
request.language
)
return {"task_id": task.id}
4.3 监控与日志方案
采用Prometheus + Grafana监控关键指标：
- 请求成功率
- 平均响应时间
- 知识检索命中率
- Token使用效率
日志记录重点捕获：
- 检索到的知识片段
- 模型输出原始结果
- 格式校验错误
5. 实战问题排查指南
5.1 输出格式不一致问题
典型症状：报告缺失某些部分或格式混乱
解决方案：
- 在Prompt中加入格式示例
- 实现输出验证器：
python复制class ReportValidator:
REQUIRED_SECTIONS = [
"安全性审查",
"性能审查",
"可维护性审查",
"综合结论"
]
def validate(self, report: str) -> bool:
return all(section in report for section in self.REQUIRED_SECTIONS)
- 设置自动重试机制
5.2 知识检索不精准问题
优化方法：
- 改进分块策略：
python复制class CodeAwareTextSplitter:
def split(self, text):
# 保持代码块完整
# 保留import/function等关键结构
- 增强元数据：
markdown复制[安全][SQL注入] 避免直接拼接SQL查询
---
风险等级: 高危
适用语言: python, java
- 采用混合检索：
python复制retriever = EnsembleRetriever(
retrievers=[
(vectorstore.as_retriever(), 0.7),
(keyword_retriever, 0.3)
]
)
5.3 长代码处理策略
对于超过500行的代码文件：
- 先进行模块分解
- 重点分析变更部分
- 全局问题扫描采用抽样检查
实现代码：
python复制def split_large_file(code):
# 按函数/类分割
# 识别变更区块
# 生成分析优先级
6. 扩展应用场景
6.1 合同审查Skill实现要点
-
知识库构建：
- 法律法规
- 历史诉讼案例
- 公司红线条款
-
审查维度：
- 权利义务对等性
- 违约责任条款
- 知识产权约定
6.2 运维诊断Skill设计
yaml复制# SKILL.md
name: "incident-diagnosis"
inputs:
- logs: List[str]
- metrics: Dict
outputs:
- root_cause: str
- severity: str
- action_items: List[str]
关键技术：
- 日志模式识别
- 指标关联分析
- 故障知识图谱
7. 演进路线建议
7.1 短期优化
- 增加测试覆盖率（目标90%+）
- 实现技能版本管理
- 开发管理控制台
7.2 中期规划
- 技能市场机制
- 自动技能组合
- 在线效果评估
7.3 长期愿景
- 自主技能进化
- 跨技能协作
- 业务流自动化
在实际项目中，我们团队采用这套架构后，业务需求响应速度提升了3倍，关键任务自动化率从15%提高到68%。最令人惊喜的是，法务部门基于合同审查Skill模板，仅用两天就自主开发出了合规检查Skill，这验证了架构的可扩展性。