知识库多轮问答

场景

希望在多轮对话中持续利用知识库检索结果，保持回答有依据、可追溯。

方案

使用 ChromaCollection + 向量模型构建 KB
TriggerFlow 负责“检索 → 回答”流程
Agent 在每轮中读取 kb 并生成简短答案

代码

python

from agently import Agently, TriggerFlow, TriggerFlowEventData
from agently.integrations.chromadb import ChromaCollection

Agently.set_settings("prompt.add_current_time", False)
Agently.set_settings("OpenAICompatible", {
  "base_url": "http://localhost:11434/v1",
  "model": "qwen2.5:7b",
  "model_type": "chat",
})

agent = Agently.create_agent()

embedding_agent = Agently.create_agent()
embedding_agent.set_settings("OpenAICompatible", {
  "base_url": "http://localhost:11434/v1",
  "model": "qwen3-embedding:0.6b",
  "model_type": "embeddings",
})

kb = ChromaCollection(
  collection_name="agently_case_kb",
  embedding_agent=embedding_agent,
  get_or_create=True,
)

kb.add([
  {
    "id": "doc-1",
    "document": "Agently 是面向生产的 AI 应用开发框架，强调输出可控与工程化编排。",
    "metadata": {"source": "overview"},
  },
  {
    "id": "doc-2",
    "document": "TriggerFlow 通过 when-emit 信号机制进行事件驱动编排，适合复杂信号系统。",
    "metadata": {"source": "triggerflow"},
  },
  {
    "id": "doc-3",
    "document": "Instant 结构化流式解析可以让结构化结果边生成边消费。",
    "metadata": {"source": "output-control"},
  },
])

questions = [
  "Agently 的核心能力是什么？",
  "TriggerFlow 和传统 DAG 编排有什么区别？",
]

flow = TriggerFlow()

@flow.chunk
def input_questions(_: TriggerFlowEventData):
  return questions

@flow.chunk
async def retrieve(data: TriggerFlowEventData):
  question = data.value
  results = kb.query(question, top_n=2)
  return {"question": question, "kb": results}

@flow.chunk
async def answer(data: TriggerFlowEventData):
  payload = data.value
  result = await (
    agent
    .input(payload["question"])
    .info({"kb": payload["kb"]})
    .instruct("只基于 {kb} 回答，给出 1-2 句话。")
    .output({"answer": ("str", "简短回答"), "sources": [("str", "引用的 doc id")]})
    .async_start()
  )
  return {"question": payload["question"], "answer": result}

(
  flow.to(input_questions)
  .for_each(concurrency=1)
  .to(retrieve)
  .to(answer)
  .end_for_each()
  .end()
)

result = flow.start("run")
print(result)

运行结果

text

[{'question': 'Agently 的核心能力是什么？', 'answer': {'answer': 'Agently的核心能力是面向生产的AI应用开发框架，强调输出可控与工程化编排。', 'sources': ['doc-1']}}, {'question': 'TriggerFlow 和传统 DAG 编排有什么区别？', 'answer': {'answer': 'TriggerFlow 使用 when-emit 信号机制进行事件驱动编排，适合复杂信号系统；而传统 DAG 编排可能依赖于任务之间的顺序。', 'sources': ['doc-2']}}]

知识库多轮问答 ​

场景 ​

方案 ​

代码 ​

运行结果 ​

知识库多轮问答

场景

方案

代码

运行结果